Missing files to rename

test/apps/cli/unit/test_teos_cli.py

@@ -0,0 +1,229 @@
+import os
+import json
+import shutil
+import responses
+from coincurve import PrivateKey
+
+import common.cryptographer
+from common.logger import Logger
+from common.tools import compute_locator
+from common.appointment import Appointment
+from common.cryptographer import Cryptographer
+
+from common.blob import Blob
+import apps.cli.teos_cli as teos_cli
+from test.apps.cli.unit.conftest import get_random_value_hex
+
+common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=teos_cli.LOG_PREFIX)
+
+# dummy keys for the tests
+dummy_sk = PrivateKey()
+dummy_pk = dummy_sk.public_key
+another_sk = PrivateKey()
+
+
+# Replace the key in the module with a key we control for the tests
+teos_cli.teos_public_key = dummy_pk
+# Replace endpoint with dummy one
+teos_cli.teos_api_server = "https://dummy.com"
+teos_cli.teos_api_port = 12345
+teos_endpoint = "{}:{}/".format(teos_cli.teos_api_server, teos_cli.teos_api_port)
+
+dummy_appointment_request = {
+    "tx": get_random_value_hex(192),
+    "tx_id": get_random_value_hex(32),
+    "start_time": 1500,
+    "end_time": 50000,
+    "to_self_delay": 200,
+}
+
+# This is the format appointment turns into once it hits "add_appointment"
+dummy_appointment_full = {
+    "locator": compute_locator(dummy_appointment_request.get("tx_id")),
+    "start_time": dummy_appointment_request.get("start_time"),
+    "end_time": dummy_appointment_request.get("end_time"),
+    "to_self_delay": dummy_appointment_request.get("to_self_delay"),
+    "encrypted_blob": Cryptographer.encrypt(
+        Blob(dummy_appointment_request.get("tx")), dummy_appointment_request.get("tx_id")
+    ),
+}
+
+dummy_appointment = Appointment.from_dict(dummy_appointment_full)
+
+
+def load_dummy_keys(*args):
+    # return dummy_pk, dummy_sk, dummy_pk_der
+    return dummy_pk
+
+
+def get_dummy_signature(*args):
+    return Cryptographer.sign(dummy_appointment.serialize(), dummy_sk)
+
+
+def get_bad_signature(*args):
+    return Cryptographer.sign(dummy_appointment.serialize(), another_sk)
+
+
+# def test_load_keys():
+#     # Let's first create a private key and public key files
+#     private_key_file_path = "sk_test_file"
+#     public_key_file_path = "pk_test_file"
+#     with open(private_key_file_path, "wb") as f:
+#         f.write(dummy_sk.to_der())
+#     with open(public_key_file_path, "wb") as f:
+#         f.write(dummy_pk_der)
+#
+#     # Now we can test the function passing the using this files (we'll use the same pk for both)
+#     r = teos_cli.load_keys(public_key_file_path, private_key_file_path, public_key_file_path)
+#     assert isinstance(r, tuple)
+#     assert len(r) == 3
+#
+#     # If any param does not match we should get None as result
+#     assert teos_cli.load_keys(None, private_key_file_path, public_key_file_path) is None
+#     assert teos_cli.load_keys(public_key_file_path, None, public_key_file_path) is None
+#     assert teos_cli.load_keys(public_key_file_path, private_key_file_path, None) is None
+#
+#     # The same should happen if we pass a public key where a private should be, for instance
+#     assert teos_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path) is None
+#
+#     os.remove(private_key_file_path)
+#     os.remove(public_key_file_path)
+
+
+# TODO: 90-add-more-add-appointment-tests
+@responses.activate
+def test_add_appointment(monkeypatch):
+    # Simulate a request to add_appointment for dummy_appointment, make sure that the right endpoint is requested
+    # and the return value is True
+    monkeypatch.setattr(teos_cli, "load_keys", load_dummy_keys)
+
+    response = {"locator": dummy_appointment.locator, "signature": get_dummy_signature()}
+    responses.add(responses.POST, teos_endpoint, json=response, status=200)
+    result = teos_cli.add_appointment([json.dumps(dummy_appointment_request)])
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.url == teos_endpoint
+    assert result
+
+
+@responses.activate
+def test_add_appointment_with_invalid_signature(monkeypatch):
+    # Simulate a request to add_appointment for dummy_appointment, but sign with a different key,
+    # make sure that the right endpoint is requested, but the return value is False
+
+    # Make sure the test uses the bad dummy signature
+    monkeypatch.setattr(teos_cli, "load_keys", load_dummy_keys)
+
+    response = {
+        "locator": dummy_appointment.to_dict()["locator"],
+        "signature": get_bad_signature(),  # Sign with a bad key
+    }
+
+    responses.add(responses.POST, teos_endpoint, json=response, status=200)
+    result = teos_cli.add_appointment([json.dumps(dummy_appointment_request)])
+
+    assert result is False
+
+
+def test_parse_add_appointment_args():
+    # If no args are passed, function should fail.
+    appt_data = teos_cli.parse_add_appointment_args(None)
+    assert not appt_data
+
+    # If file doesn't exist, function should fail.
+    appt_data = teos_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
+    assert not appt_data
+
+    # If file exists and has data in it, function should work.
+    with open("appt_test_file", "w") as f:
+        json.dump(dummy_appointment_request, f)
+
+    appt_data = teos_cli.parse_add_appointment_args(["-f", "appt_test_file"])
+    assert appt_data
+
+    os.remove("appt_test_file")
+
+    # If appointment json is passed in, function should work.
+    appt_data = teos_cli.parse_add_appointment_args([json.dumps(dummy_appointment_request)])
+    assert appt_data
+
+
+@responses.activate
+def test_post_appointment():
+    response = {
+        "locator": dummy_appointment.to_dict()["locator"],
+        "signature": Cryptographer.sign(dummy_appointment.serialize(), dummy_pk),
+    }
+
+    responses.add(responses.POST, teos_endpoint, json=response, status=200)
+    response = teos_cli.post_appointment(json.dumps(dummy_appointment_request))
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.url == teos_endpoint
+    assert response
+
+
+@responses.activate
+def test_process_post_appointment_response():
+    # Let's first crete a response
+    response = {
+        "locator": dummy_appointment.to_dict()["locator"],
+        "signature": Cryptographer.sign(dummy_appointment.serialize(), dummy_pk),
+    }
+
+    # A 200 OK with a correct json response should return the json of the response
+    responses.add(responses.POST, teos_endpoint, json=response, status=200)
+    r = teos_cli.post_appointment(json.dumps(dummy_appointment_request))
+    assert teos_cli.process_post_appointment_response(r) == r.json()
+
+    # If we modify the response code tor a rejection (lets say 404) we should get None
+    responses.replace(responses.POST, teos_endpoint, json=response, status=404)
+    r = teos_cli.post_appointment(json.dumps(dummy_appointment_request))
+    assert teos_cli.process_post_appointment_response(r) is None
+
+    # The same should happen if the response is not in json
+    responses.replace(responses.POST, teos_endpoint, status=404)
+    r = teos_cli.post_appointment(json.dumps(dummy_appointment_request))
+    assert teos_cli.process_post_appointment_response(r) is None
+
+
+def test_save_appointment_receipt(monkeypatch):
+    appointments_folder = "test_appointments_receipts"
+    teos_cli.config["APPOINTMENTS_FOLDER_NAME"] = appointments_folder
+
+    # The functions creates a new directory if it does not exist
+    assert not os.path.exists(appointments_folder)
+    teos_cli.save_appointment_receipt(dummy_appointment.to_dict(), get_dummy_signature())
+    assert os.path.exists(appointments_folder)
+
+    # Check that the receipt has been saved by checking the file names
+    files = os.listdir(appointments_folder)
+    assert any([dummy_appointment.locator in f for f in files])
+
+    shutil.rmtree(appointments_folder)
+
+
+@responses.activate
+def test_get_appointment():
+    # Response of get_appointment endpoint is an appointment with status added to it.
+    dummy_appointment_full["status"] = "being_watched"
+    response = dummy_appointment_full
+
+    request_url = "{}get_appointment?locator={}".format(teos_endpoint, response.get("locator"))
+    responses.add(responses.GET, request_url, json=response, status=200)
+    result = teos_cli.get_appointment(response.get("locator"))
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.url == request_url
+    assert result.get("locator") == response.get("locator")
+
+
+@responses.activate
+def test_get_appointment_err():
+    locator = get_random_value_hex(16)
+
+    # Test that get_appointment handles a connection error appropriately.
+    request_url = "{}get_appointment?locator=".format(teos_endpoint, locator)
+    responses.add(responses.GET, request_url, body=ConnectionError())
+
+    assert not teos_cli.get_appointment(locator)