Reputation: 169
Here's what I am doing:
def _wait_for_long_poll_resp(param_one, q: Queue):
print("long poll called", flush=True)
sys.stdout.flush()
request_url = <some_url>
auth = <auth>
while True:
print(f'I am Calling {request_url}', flush=True)
sys.stdout.flush()
resp = requests.get(request_url, headers=HEADERS, auth=auth)
print(resp.status_code, flush=True)
sys.stdout.flush()
print(resp.content, flush=True)
sys.stdout.flush()
if resp.status_code == 200:
print("Response code is 200", flush=True)
print(resp.content, flush=True)
long_poll_resp = LongPollResponse(
status_code=resp.status_code,
content=resp.content
)
else:
print("Response code is not 200", flush=True)
print(resp.status_code, flush=True)
long_poll_resp = LongPollResponse(status_code=resp.status_code)
q.put(long_poll_resp)
@contextmanager
def long_pool_listener_for_charger(param_one):
q = Queue()
p = Process(target=_wait_for_long_poll_resp, args=[charge_point, q])
try:
print("In try block")
p.start()
sleep(2.0)
yield q
except Exception as e:
raise Exception("Error occurred: ", e)
finally:
p.join()
@contextmanager
def long_pool_listener(param_one):
with long_pool_listener_for_charger(param_one) as listener:
yield listener
And then I am writing a test for this using pytest:
def test_remote_reset():
param_one = "something"
with long_pool_listener(param_one) as q: # type: Queue
sleep(3)
request_url = <some_url>
headers = <some_headers>
auth = <auth>
print(f'Calling {request_url}')
resp = requests.post(request_url, headers=headers, auth=auth)
# Give it some time
sleep(3)
resp = q.get(timeout=15)
# Some assertions
Is this code correct? Because it doesn't seem to even enter the function, _wait_for_long_poll_resp. I know this because it's not even printing the debug prints I have written here. Also wanted to know if it's pythonic enough.
Upvotes: 0
Views: 93