mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 22:24:23 +01:00
Changed to_json to to_dict in Appointment and Job; added to_json to Appointment to actually return a string
This commit is contained in:
1
.gitignore
vendored
1
.gitignore
vendored
@@ -12,3 +12,4 @@ appointments/
|
|||||||
test.py
|
test.py
|
||||||
*.pyc
|
*.pyc
|
||||||
.cache
|
.cache
|
||||||
|
.pytest_cache/
|
||||||
@@ -68,7 +68,7 @@ def get_appointment():
|
|||||||
|
|
||||||
if appointment_in_watcher:
|
if appointment_in_watcher:
|
||||||
for uuid in appointment_in_watcher:
|
for uuid in appointment_in_watcher:
|
||||||
appointment_data = watcher.appointments[uuid].to_json()
|
appointment_data = watcher.appointments[uuid].to_dict()
|
||||||
appointment_data['status'] = "being_watched"
|
appointment_data['status'] = "being_watched"
|
||||||
response.append(appointment_data)
|
response.append(appointment_data)
|
||||||
|
|
||||||
@@ -77,7 +77,7 @@ def get_appointment():
|
|||||||
|
|
||||||
for job in responder_jobs.values():
|
for job in responder_jobs.values():
|
||||||
if job.locator == locator:
|
if job.locator == locator:
|
||||||
job_data = job.to_json()
|
job_data = job.to_dict()
|
||||||
job_data['status'] = "dispute_responded"
|
job_data['status'] = "dispute_responded"
|
||||||
response.append(job_data)
|
response.append(job_data)
|
||||||
|
|
||||||
@@ -98,11 +98,11 @@ def get_all_appointments():
|
|||||||
|
|
||||||
if request.remote_addr in request.host or request.remote_addr == '127.0.0.1':
|
if request.remote_addr in request.host or request.remote_addr == '127.0.0.1':
|
||||||
for uuid, appointment in watcher.appointments.items():
|
for uuid, appointment in watcher.appointments.items():
|
||||||
watcher_appointments[uuid] = appointment.to_json()
|
watcher_appointments[uuid] = appointment.to_dict()
|
||||||
|
|
||||||
if watcher.responder:
|
if watcher.responder:
|
||||||
for uuid, job in watcher.responder.jobs.items():
|
for uuid, job in watcher.responder.jobs.items():
|
||||||
responder_jobs[uuid] = job.to_json()
|
responder_jobs[uuid] = job.to_dict()
|
||||||
|
|
||||||
response = jsonify({"watcher_appointments": watcher_appointments, "responder_jobs": responder_jobs})
|
response = jsonify({"watcher_appointments": watcher_appointments, "responder_jobs": responder_jobs})
|
||||||
|
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ class Appointment:
|
|||||||
self.cipher = cipher
|
self.cipher = cipher
|
||||||
self.hash_function = hash_function
|
self.hash_function = hash_function
|
||||||
|
|
||||||
def to_json(self):
|
def to_dict(self):
|
||||||
appointment = {"locator": self.locator, "start_time": self.start_time, "end_time": self.end_time,
|
appointment = {"locator": self.locator, "start_time": self.start_time, "end_time": self.end_time,
|
||||||
"dispute_delta": self.dispute_delta, "encrypted_blob": self.encrypted_blob.data,
|
"dispute_delta": self.dispute_delta, "encrypted_blob": self.encrypted_blob.data,
|
||||||
"cipher": self.cipher, "hash_function": self.hash_function}
|
"cipher": self.cipher, "hash_function": self.hash_function}
|
||||||
@@ -24,5 +24,5 @@ class Appointment:
|
|||||||
|
|
||||||
# ToDO: #3-improve-appointment-structure
|
# ToDO: #3-improve-appointment-structure
|
||||||
|
|
||||||
def serialize(self):
|
def to_json(self):
|
||||||
return json.dumps(self.to_json())
|
return json.dumps(self.to_dict())
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ class Job:
|
|||||||
# can be directly got from DB
|
# can be directly got from DB
|
||||||
self.locator = sha256(unhexlify(dispute_txid)).hexdigest()
|
self.locator = sha256(unhexlify(dispute_txid)).hexdigest()
|
||||||
|
|
||||||
def to_json(self):
|
def to_dict(self):
|
||||||
job = {"locator": self.locator, "justice_rawtx": self.justice_rawtx, "appointment_end": self.appointment_end}
|
job = {"locator": self.locator, "justice_rawtx": self.justice_rawtx, "appointment_end": self.appointment_end}
|
||||||
|
|
||||||
return job
|
return job
|
||||||
|
|||||||
@@ -64,7 +64,7 @@ class Watcher:
|
|||||||
logger.info("New appointment accepted.", locator=appointment.locator)
|
logger.info("New appointment accepted.", locator=appointment.locator)
|
||||||
|
|
||||||
if (self.signing_key is not None):
|
if (self.signing_key is not None):
|
||||||
signature = self.signing_key.sign(appointment.serialize().encode('utf8'))
|
signature = self.signing_key.sign(appointment.to_json().encode('utf8'))
|
||||||
|
|
||||||
else:
|
else:
|
||||||
appointment_added = False
|
appointment_added = False
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
from os import urandom
|
from os import urandom
|
||||||
|
import json
|
||||||
from pytest import fixture
|
from pytest import fixture
|
||||||
|
|
||||||
from pisa.appointment import Appointment
|
from pisa.appointment import Appointment
|
||||||
@@ -35,13 +36,25 @@ def test_init_appointment(appointment_data):
|
|||||||
and dispute_delta == appointment.dispute_delta and hash_function == appointment.hash_function)
|
and dispute_delta == appointment.dispute_delta and hash_function == appointment.hash_function)
|
||||||
|
|
||||||
|
|
||||||
|
def test_to_dict(appointment_data):
|
||||||
|
locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function = appointment_data
|
||||||
|
appointment = Appointment(locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function)
|
||||||
|
|
||||||
|
dict_appointment = appointment.to_dict()
|
||||||
|
|
||||||
|
assert (locator == dict_appointment.get("locator") and start_time == dict_appointment.get("start_time")
|
||||||
|
and end_time == dict_appointment.get("end_time") and dispute_delta == dict_appointment.get("dispute_delta")
|
||||||
|
and cipher == dict_appointment.get("cipher") and hash_function == dict_appointment.get("hash_function")
|
||||||
|
and encrypted_blob_data == dict_appointment.get("encrypted_blob"))
|
||||||
|
|
||||||
|
|
||||||
def test_to_json(appointment_data):
|
def test_to_json(appointment_data):
|
||||||
locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function = appointment_data
|
locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function = appointment_data
|
||||||
appointment = Appointment(locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function)
|
appointment = Appointment(locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function)
|
||||||
|
|
||||||
json_appointment = appointment.to_json()
|
dict_appointment = json.loads(appointment.to_json())
|
||||||
|
|
||||||
assert (locator == json_appointment.get("locator") and start_time == json_appointment.get("start_time")
|
assert (locator == dict_appointment.get("locator") and start_time == dict_appointment.get("start_time")
|
||||||
and end_time == json_appointment.get("end_time") and dispute_delta == json_appointment.get("dispute_delta")
|
and end_time == dict_appointment.get("end_time") and dispute_delta == dict_appointment.get("dispute_delta")
|
||||||
and cipher == json_appointment.get("cipher") and hash_function == json_appointment.get("hash_function")
|
and cipher == dict_appointment.get("cipher") and hash_function == dict_appointment.get("hash_function")
|
||||||
and encrypted_blob_data == json_appointment.get("encrypted_blob"))
|
and encrypted_blob_data == dict_appointment.get("encrypted_blob"))
|
||||||
|
|||||||
Reference in New Issue
Block a user