Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_cargo_truck_init():
cgo = CargoTruck(['a', 'b'], 2)
assert len(cgo.orders) == 2
assert cgo.orders[0] == 'a'
assert cgo.num_loaders == 2
print("at the end of the rope")
items_remaining = self._servable - self._served
num_threads = self.max_threads
max_chunksize = self.pageSize * num_threads
if items_remaining - max_chunksize < 0:
full, leftovers = divmod(items_remaining, self.pageSize)
num_threads = full + (1 if leftovers else 0)
stixes = [i+1 for i in range(num_threads)]
page_urls = [re.sub('&start=\d+', '&start=%s' % (self.startIndex + (i * self.pageSize)), self.resource)
for i in stixes]
success = False
exc = None
delays = [0, 2, 5]
for delay in delays:
time.sleep(delay)
cgt = CargoTruck(page_urls, num_threads)
try:
cgt.load(self.session, 'get', 15)
payload = cgt.dump()
success = True
break
except:
exc_name, exc_desc = sys.exc_info()[:2]
anomaly = "RallyResponse.next.__retrieveNextPage.__retrievePages caught exception " +\
"in threaded request/response activity: %s, %s" % (exc_name, exc_desc)
pg1, pg2 = (self.startIndex + self.pageSize), (self.startIndex + (self.pageSize*num_threads))
notice = "Retrying the page_urls for the page group startIndexes: %d -> %d" % (pg1, pg2)
print(notice)
if not success:
raise RallyResponseError("Unable to retrieve %d chunks of data" % num_threads)