aljaz/python-teos
1
0
Fork 0
mirror of https://github.com/aljazceru/python-teos.git synced 2025-12-17 06:04:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fixes default HOST to remove the schema from config

Browse Source 
The schema now defaults to http if none is defined. Also improves some of the cli docstrings
This commit is contained in:
Sergi Delgado Segura
2020-03-23 15:09:24 +01:00
parent 02eefd5807
commit f5edaf1bf0
4 changed files with 79 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cli/__init__.py
View File

@@ -6,7 +6,7 @@ LOG_PREFIX = "cli"
# Load config fields
DEFAULT_CONF = {
"TEOS_SERVER": {"value": "http://localhost", "type": str},
"TEOS_SERVER": {"value": "localhost", "type": str},
"TEOS_PORT": {"value": 9814, "type": int},
"LOG_FILE": {"value": "teos_cli.log", "type": str, "path": True},
"APPOINTMENTS_FOLDER_NAME": {"value": "appointment_receipts", "type": str, "path": True},

2
cli/template.conf
View File

@@ -1,4 +1,4 @@
[teos]
TEOS_SERVER = "http://localhost"
TEOS_SERVER = "localhost"
TEOS_PORT = 9814

110
cli/teos_cli.py
View File

@@ -1,14 +1,15 @@
import os
import sys
import time
import json
import requests
import time
import binascii
from sys import argv
from uuid import uuid4
from coincurve import PublicKey
from getopt import getopt, GetoptError
from requests import ConnectTimeout, ConnectionError
from requests.exceptions import MissingSchema, InvalidSchema, InvalidURL
from cli.help import help_add_appointment, help_get_appointment
from cli import DEFAULT_CONF, DATA_DIR, CONF_FILE_NAME, LOG_PREFIX
@@ -79,7 +80,7 @@ def load_keys(teos_pk_path, cli_sk_path, cli_pk_path):
return teos_pk, cli_sk, cli_pk_der
def add_appointment(args, config):
def add_appointment(args, teos_url, config):
"""
Manages the add_appointment command, from argument parsing, trough sending the appointment to the tower, until
saving the appointment receipt.
@@ -100,12 +101,17 @@ def add_appointment(args, config):
Args:
args (:obj:`list`): a list of arguments to pass to ``parse_add_appointment_args``. Must contain a json encoded
appointment, or the file option and the path to a file containing a json encoded appointment.
teos_url (:obj:`str`): the teos base url.
config (:obj:`dict`): a config dictionary following the format of :func:`create_config_dict <common.config_loader.ConfigLoader.create_config_dict>`.
Returns:
:obj:`bool`: True if the appointment is accepted by the tower and the receipt is properly stored, false if any
error occurs during the process.
"""
# Currently the base_url is the same as the add_appointment_endpoint
add_appointment_endpoint = teos_url
teos_pk, cli_sk, cli_pk_der = load_keys(
config.get("TEOS_PUBLIC_KEY"), config.get("CLI_PRIVATE_KEY"), config.get("CLI_PUBLIC_KEY")
)
@@ -153,7 +159,7 @@ def add_appointment(args, config):
data = {"appointment": appointment.to_dict(), "signature": signature, "public_key": hex_pk_der.decode("utf-8")}
# Send appointment to the server.
server_response = post_appointment(data, config)
server_response = post_appointment(data, add_appointment_endpoint)
if server_response is None:
return False
@@ -227,13 +233,14 @@ def parse_add_appointment_args(args):
return appointment_data
def post_appointment(data, config):
def post_appointment(data, add_appointment_endpoint):
"""
Sends appointment data to add_appointment endpoint to be processed by the tower.
Args:
data (:obj:`dict`): a dictionary containing three fields: an appointment, the client-side signature, and the
der-encoded client public key.
add_appointment_endpoint (:obj:`str`): the teos endpoint where to send appointments to.
Returns:
:obj:`dict` or ``None``: a json-encoded dictionary with the server response if the data can be posted.
@@ -243,7 +250,6 @@ def post_appointment(data, config):
logger.info("Sending appointment to the Eye of Satoshi")
try:
add_appointment_endpoint = "{}:{}".format(config.get("TEOS_SERVER"), config.get("TEOS_PORT"))
return requests.post(url=add_appointment_endpoint, json=json.dumps(data), timeout=5)
except ConnectTimeout:
@@ -254,8 +260,8 @@ def post_appointment(data, config):
logger.error("Can't connect to the Eye of Satoshi's API. Server cannot be reached")
return None
except requests.exceptions.InvalidSchema:
logger.error("No transport protocol found. Have you missed http(s):// in the server url?")
except (InvalidSchema, MissingSchema, InvalidURL):
logger.error("Invalid URL. No schema, or invalid schema, found ({})".format(add_appointment_endpoint))
except requests.exceptions.Timeout:
logger.error("The request timed out")
@@ -266,7 +272,7 @@ def process_post_appointment_response(response):
Processes the server response to an add_appointment request.
Args:
response (:obj:`requests.models.Response`): a ``Response` object obtained from the sent request.
response (:obj:`requests.models.Response`): a ``Response`` object obtained from the sent request.
Returns:
:obj:`dict` or :obj:`None`: a dictionary containing the tower's response data if it can be properly parsed and
@@ -306,6 +312,7 @@ def save_appointment_receipt(appointment, signature, config):
Args:
appointment (:obj:`Appointment <common.appointment.Appointment>`): the appointment to be saved on disk.
signature (:obj:`str`): the signature of the appointment performed by the tower.
config (:obj:`dict`): a config dictionary following the format of :func:`create_config_dict <common.config_loader.ConfigLoader.create_config_dict>`.
Returns:
:obj:`bool`: True if the appointment if properly saved, false otherwise.
@@ -335,12 +342,13 @@ def save_appointment_receipt(appointment, signature, config):
return False
def get_appointment(locator, config):
def get_appointment(locator, get_appointment_endpoint):
"""
Gets information about an appointment from the tower.
Args:
locator (:obj:`str`): the appointment locator used to identify it.
get_appointment_endpoint (:obj:`str`): the teos endpoint where to get appointments from.
Returns:
:obj:`dict` or :obj:`None`: a dictionary containing thew appointment data if the locator is valid and the tower
@@ -353,7 +361,6 @@ def get_appointment(locator, config):
logger.error("The provided locator is not valid", locator=locator)
return None
get_appointment_endpoint = "{}:{}/get_appointment".format(config.get("TEOS_SERVER"), config.get("TEOS_PORT"))
parameters = "?locator={}".format(locator)
try:
@@ -384,49 +391,34 @@ def show_usage():
"\n\tget_appointment \tGets json formatted data about an appointment from the tower."
"\n\thelp \t\t\tShows a list of commands or help for a specific command."
"\n\nGLOBAL OPTIONS:"
"\n\t-s, --server \tAPI server where to send the requests. Defaults to https://teos.pisa.watch (modifiable in "
"config.py)"
"\n\t-p, --port \tAPI port where to send the requests. Defaults to 443 (modifiable in conf.py)"
"\n\t-d, --debug \tshows debug information and stores it in teos_cli.log"
"\n\t-s, --server \tAPI server where to send the requests. Defaults to localhost (modifiable in conf file)."
"\n\t-p, --port \tAPI port where to send the requests. Defaults to 9814 (modifiable in conf file)."
"\n\t-d, --debug \tshows debug information and stores it in teos_cli.log."
"\n\t-h --help \tshows this message."
)
if __name__ == "__main__":
command_line_conf = {}
commands = ["add_appointment", "get_appointment", "help"]
def main(args, command_line_conf):
# Loads config and sets up the data folder and log file
config_loader = ConfigLoader(DATA_DIR, CONF_FILE_NAME, DEFAULT_CONF, command_line_conf)
config = config_loader.build_config()
setup_data_folder(DATA_DIR)
setup_logging(config.get("LOG_FILE"), LOG_PREFIX)
# Set the teos url
teos_url = "{}:{}".format(config.get("TEOS_SERVER"), config.get("TEOS_PORT"))
# If an http or https prefix if found, leaves the server as is. Otherwise defaults to http.
if not teos_url.startswith("http"):
teos_url = "http://" + teos_url
try:
opts, args = getopt(argv[1:], "s:p:h", ["server", "port", "help"])
for opt, arg in opts:
if opt in ["-s", "--server"]:
if arg:
command_line_conf["TEOS_SERVER"] = arg
if opt in ["-p", "--port"]:
if arg:
try:
command_line_conf["TEOS_PORT"] = int(arg)
except ValueError:
exit("port must be an integer")
if opt in ["-h", "--help"]:
sys.exit(show_usage())
# Loads config and sets up the data folder and log file
config_loader = ConfigLoader(DATA_DIR, CONF_FILE_NAME, DEFAULT_CONF, command_line_conf)
config = config_loader.build_config()
setup_data_folder(DATA_DIR)
setup_logging(config.get("LOG_FILE"), LOG_PREFIX)
if args:
command = args.pop(0)
if command in commands:
if command == "add_appointment":
add_appointment(args, config)
add_appointment(args, teos_url, config)
elif command == "get_appointment":
if not args:
@@ -438,7 +430,8 @@ if __name__ == "__main__":
if arg_opt in ["-h", "--help"]:
sys.exit(help_get_appointment())
appointment_data = get_appointment(arg_opt, config)
get_appointment_endpoint = "{}/get_appointment".format(teos_url)
appointment_data = get_appointment(arg_opt, get_appointment_endpoint)
if appointment_data:
print(appointment_data)
@@ -464,8 +457,33 @@ if __name__ == "__main__":
else:
logger.error("No command provided. Use help to check the list of available commands")
except json.JSONDecodeError:
logger.error("Non-JSON encoded appointment passed as parameter")
if __name__ == "__main__":
command_line_conf = {}
commands = ["add_appointment", "get_appointment", "help"]
try:
opts, args = getopt(argv[1:], "s:p:h", ["server", "port", "help"])
for opt, arg in opts:
if opt in ["-s", "--server"]:
if arg:
command_line_conf["TEOS_SERVER"] = arg
if opt in ["-p", "--port"]:
if arg:
try:
command_line_conf["TEOS_PORT"] = int(arg)
except ValueError:
exit("port must be an integer")
if opt in ["-h", "--help"]:
sys.exit(show_usage())
main(args, command_line_conf)
except GetoptError as e:
logger.error("{}".format(e))
except json.JSONDecodeError as e:
logger.error("Non-JSON encoded appointment passed as parameter")

24
test/cli/unit/test_teos_cli.py
View File

@@ -27,7 +27,7 @@ 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_endpoint = "{}:{}/".format(config.get("TEOS_SERVER"), config.get("TEOS_PORT"))
teos_endpoint = "http://{}:{}/".format(config.get("TEOS_SERVER"), config.get("TEOS_PORT"))
dummy_appointment_request = {
"tx": get_random_value_hex(192),
@@ -107,7 +107,7 @@ def test_add_appointment(monkeypatch):
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)], config)
result = teos_cli.add_appointment([json.dumps(dummy_appointment_request)], teos_endpoint, config)
assert len(responses.calls) == 1
assert responses.calls[0].request.url == teos_endpoint
@@ -128,7 +128,7 @@ def test_add_appointment_with_invalid_signature(monkeypatch):
}
responses.add(responses.POST, teos_endpoint, json=response, status=200)
result = teos_cli.add_appointment([json.dumps(dummy_appointment_request)], config)
result = teos_cli.add_appointment([json.dumps(dummy_appointment_request)], teos_endpoint, config)
shutil.rmtree(config.get("APPOINTMENTS_FOLDER_NAME"))
@@ -166,7 +166,7 @@ def test_post_appointment():
}
responses.add(responses.POST, teos_endpoint, json=response, status=200)
response = teos_cli.post_appointment(json.dumps(dummy_appointment_request), config)
response = teos_cli.post_appointment(json.dumps(dummy_appointment_request), teos_endpoint)
assert len(responses.calls) == 1
assert responses.calls[0].request.url == teos_endpoint
@@ -183,17 +183,17 @@ def test_process_post_appointment_response():
# 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), config)
r = teos_cli.post_appointment(json.dumps(dummy_appointment_request), teos_endpoint)
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), config)
r = teos_cli.post_appointment(json.dumps(dummy_appointment_request), teos_endpoint)
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), config)
r = teos_cli.post_appointment(json.dumps(dummy_appointment_request), teos_endpoint)
assert teos_cli.process_post_appointment_response(r) is None
@@ -218,10 +218,11 @@ 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
get_appointment_endpoint = teos_endpoint + "get_appointment"
request_url = "{}get_appointment?locator={}".format(teos_endpoint, response.get("locator"))
request_url = "{}?locator={}".format(get_appointment_endpoint, response.get("locator"))
responses.add(responses.GET, request_url, json=response, status=200)
result = teos_cli.get_appointment(response.get("locator"), config)
result = teos_cli.get_appointment(response.get("locator"), get_appointment_endpoint)
assert len(responses.calls) == 1
assert responses.calls[0].request.url == request_url
@@ -231,9 +232,10 @@ def test_get_appointment():
@responses.activate
def test_get_appointment_err():
locator = get_random_value_hex(16)
get_appointment_endpoint = teos_endpoint + "get_appointment"
# Test that get_appointment handles a connection error appropriately.
request_url = "{}get_appointment?locator=".format(teos_endpoint, locator)
request_url = "{}?locator=".format(get_appointment_endpoint, locator)
responses.add(responses.GET, request_url, body=ConnectionError())
assert not teos_cli.get_appointment(locator, config)
assert not teos_cli.get_appointment(locator, get_appointment_endpoint)