Commit 3edd5e25 authored by Sli's avatar Sli
Browse files

Merge branch 'testing'

parents 93c2e264 6e6a7896
......@@ -23,3 +23,5 @@ swagger-client
*_pb2*
.DS_Store
.envrc
pyrightconfig.json
*.egg-info*
\ No newline at end of file
server/settings_custom.py
settings_custom.py
setup.py
poetry.lock
*.sqlite
*.pyc
\ No newline at end of file
# Installation
```bash
virtualenv python=python3 env
source env/bin/activate
pip install -r requirements.txt
poetry install
poetry shell
./manage.py protoc # Generate protobuf files
./manage.py setup --import 2019.json # Generate database
./manage.py runserver -h 127.0.0.1
cash setup 2019.json # Generate database by importing 2019's database
cash runserver
# You can use custom settings by creating a settings_custom.py file
# You can also configure the location of those settings through the CFG env variable
```
# Testing
```bash
cash test # Launch all tests
cash test TestBalance # Only launch test cases of TestBalance class
cash test TestBalance.test_existing_user # Only test one method of TestBalance class
```
# Developement
```bash
# Rebuild protobuf files
cash protoc
```
import sys
import os
from shutil import copyfile
from subprocess import Popen
import logging
def build(_):
logging.basicConfig(level=logging.INFO)
try:
Popen(f"{sys.executable} ./protoc.py", shell=True).wait()
except:
logging.warning("Could not generate prtobuf files. This is normal if you are installing from release package")
# -*- coding:utf-8 -*
import logging
logging.basicConfig(level=logging.INFO)
from . import settings
from sqlalchemy import create_engine
from sqlalchemy.orm.session import Session
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine("sqlite:///%s" % settings.DB_PATH, convert_unicode=True)
db = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))
db: Session = scoped_session(
sessionmaker(autocommit=False, autoflush=False, bind=engine)
)
Model = declarative_base()
Model.query = db.query_property()
......@@ -4,6 +4,7 @@ import os
import sys
import click
import logging
@click.group()
......@@ -13,7 +14,11 @@ def default_group():
@default_group.command(name="runserver", help="Run the server")
@click.option(
"--host", "-h", default="", type=str, help='Host to listen from (default "")'
"--host",
"-h",
default="127.0.0.1",
type=str,
help='Host to listen from (default "")',
)
@click.option(
"--port", "-p", default=50051, type=int, help="Port to listen from (default 50051)"
......@@ -26,47 +31,47 @@ def default_group():
help="Enable server reflection for debug (default False)",
)
def runserver(host, port, reflect):
from server import server
from . import server
server.serve(host, port, reflect)
@default_group.command(name="test", help="Run tests")
@click.argument("name", type=str, required=False)
def test(name=None):
import unittest
loader = unittest.TestLoader()
if name is None:
suite = loader.discover("", pattern="tests.py")
else:
suite = loader.loadTestsFromName("server.tests.%s" % name)
unittest.TextTestRunner(verbosity=2).run(suite)
@default_group.command(name="protoc", help="Generate protoc files")
def protoc():
from subprocess import Popen
import fileinput
p = Popen(
"python -m grpc_tools.protoc -I%s --python_out=%s --grpc_python_out=%s %s"
% ("../protos", "./server/", "./server/", "../protos/com.proto"),
shell=True,
)
p.wait()
# Fix import path issues
with fileinput.FileInput("server/com_pb2_grpc.py", inplace=True) as file:
for line in file:
print(
line.replace("import com_pb2", "import server.com_pb2"), end="",
)
try:
from ..protoc import build_protoc
except ModuleNotFoundError:
logging.error("This functionnality can not be used when installed from pip")
return
build_protoc()
@default_group.command(name="setup", help="Generate database")
@click.option("--import", "-i", default=None, help="Import initial data from json file")
@click.option(
"--schema",
"-s",
default="db_import_schema.json",
help="Location of the schema used to validate the json of the import command",
)
def setup(**kwargs):
@click.argument("import_file", required=False, default=None, type=click.File("r"))
def setup(import_file):
import json
from importlib.resources import read_text
from datetime import datetime
from random import randint
from slugify import slugify
import jsonschema
from server import settings, db, Model, engine, com_pb2, models
from . import settings, db, Model, engine, com_pb2, models
def datetime_helper(event_date, hour, minute):
# Does not handle midnight and after
......@@ -86,13 +91,13 @@ def setup(**kwargs):
return code
if os.path.exists(settings.DB_PATH):
print("--- Deleting database ---")
logging.info("--- Deleting database ---")
os.remove(settings.DB_PATH)
print("--- Creating database ---")
logging.info("--- Creating database ---")
Model.metadata.create_all(bind=engine)
print("--- Creating payment methods ---")
logging.info("--- Creating payment methods ---")
db.add(models.PaymentMethod(id=com_pb2.PaymentMethod.UNKNOWN, name="inconnu"))
db.add(models.PaymentMethod(id=com_pb2.PaymentMethod.CASH, name="espèces"))
db.add(models.PaymentMethod(id=com_pb2.PaymentMethod.CARD, name="carte"))
......@@ -101,40 +106,24 @@ def setup(**kwargs):
db.add(models.PaymentMethod(id=com_pb2.PaymentMethod.TRANSFER, name="transfert"))
db.add(models.PaymentMethod(id=com_pb2.PaymentMethod.OTHER, name="autre"))
if kwargs["import"] is not None:
if import_file is not None:
# Load user json
print("Importing data from %s" % kwargs["import"])
try:
f = open(kwargs["import"], "r")
except OSError as e:
print("Error opening %s: %s" % (kwargs["import"], e), file=sys.stderr)
return
logging.info(f"Importing data from {import_file.name}")
try:
data = json.load(f)
data = json.loads(import_file.read())
except json.decoder.JSONDecodeError as e:
print("Error loading %s: %s" % (kwargs["import"], e), file=sys.stderr)
logging.error(f"Error loading {import_file.name}: {e}")
return
f.close()
print(kwargs["schema"])
import_file.close()
# Load schema for validation
print("Matching against schema")
try:
f = open(kwargs["schema"], "r")
except OSError as e:
print("Error opening %s: %s" % (kwargs["import"], e), file=sys.stderr)
return
try:
schema = json.load(f)
except json.decoder.JSONDecodeError as e:
print("Error loading %s: %s" % (kwargs["import"], e), file=sys.stderr)
return
f.close()
# Load schema for validation from current module resources files
schema = json.loads(read_text(__package__, "db_import_schema.json"))
logging.info("Matching against schema")
try:
jsonschema.validate(data, schema)
except jsonschema.exceptions.ValidationError as e:
print("Your json is incorrect: %s" % e, file=sys.stderr)
logging.error(f"Your json is incorrect: {e}")
return
generated_codes = {}
......@@ -142,16 +131,16 @@ def setup(**kwargs):
event_date = datetime(
year=event_date["year"], month=event_date["month"], day=event_date["day"]
)
print("--- Event date is : %s ---", event_date)
logging.info(f"--- Event date is : {event_date} ---")
print("--- Creating counters ---")
logging.info("--- Creating counters ---")
for counter in data.get("counters", []):
print("Creating counter %s" % counter)
logging.info(f"Creating counter {counter}")
db.add(models.Counter(name=counter))
db.commit()
print("--- Creating products ---")
logging.info("--- Creating products ---")
for product in data.get("products", []):
p = models.Product(
name=product["name"].capitalize(),
......@@ -161,7 +150,7 @@ def setup(**kwargs):
),
default_price=product["price"],
)
print("Creating product %s" % p.name)
logging.info(f"Creating product {p.name}")
db.add(p)
db.commit()
......@@ -196,8 +185,12 @@ def setup(**kwargs):
db.commit()
db.commit()
print("--- Database successfully created ---")
logging.info("--- Database successfully created ---")
if __name__ == "__main__":
def main():
click.CommandCollection(sources=[default_group])()
if __name__ == "__main__":
main()
......@@ -13,7 +13,7 @@ from sqlalchemy import (
from sqlalchemy.orm import relationship, backref
import sqlalchemy.types as types
from server import Model
from . import Model
class Money(types.TypeDecorator):
......
#!/usr/bin/env python3
# -*- coding:utf-8 -*
from google.protobuf.timestamp_pb2 import Timestamp
import decimal
from . import models, com_pb2
def pb_now() -> Timestamp:
timestamp = Timestamp()
timestamp.GetCurrentTime()
return timestamp
def date_to_pb(date) -> Timestamp:
timestamp = Timestamp()
timestamp.FromDatetime(date)
return timestamp
def decimal_to_pb_money(dec: decimal.Decimal) -> com_pb2.Money:
return com_pb2.Money(amount=str(dec))
def pb_money_to_decimal(money: com_pb2.Money) -> decimal.Decimal:
try:
return decimal.Decimal(money.amount)
except:
return decimal.Decimal(0)
def refilling_to_pb(refilling: models.Refilling) -> com_pb2.Refilling:
return com_pb2.Refilling(
id=refilling.id,
customer_id=refilling.customer_id,
counter_id=refilling.counter_id,
device_uuid=refilling.machine_id,
payment_method=refilling.payment_method.id,
amount=decimal_to_pb_money(refilling.amount),
cancelled=refilling.cancelled,
date=date_to_pb(refilling.date),
)
from subprocess import Popen
import sys
import fileinput
def build_protoc():
p = Popen(
"%s -m grpc_tools.protoc -I%s --python_out=%s --grpc_python_out=%s %s"
% (sys.executable, "../protos", "./cashless_server/", "./cashless_server/", "../protos/com.proto"),
shell=True,
)
p.wait()
# Fix import path issues
with fileinput.FileInput("cashless_server/com_pb2_grpc.py", inplace=True) as file:
for line in file:
print(
line.replace("import com_pb2", "import cashless_server.com_pb2"), end="",
)
if __name__ == "__main__":
build_protoc()
\ No newline at end of file
......@@ -4,47 +4,19 @@
import grpc
import json
import decimal
import logging
from concurrent import futures
from google.protobuf.timestamp_pb2 import Timestamp
from server import db, models, com_pb2, com_pb2_grpc
def pb_now() -> Timestamp:
timestamp = Timestamp()
timestamp.GetCurrentTime()
return timestamp
def date_to_pb(date) -> Timestamp:
timestamp = Timestamp()
timestamp.FromDatetime(date)
return timestamp
def decimal_to_pb_money(dec: decimal.Decimal) -> com_pb2.Money:
return com_pb2.Money(amount=str(dec))
def pb_money_to_decimal(money: com_pb2.Money) -> decimal.Decimal:
try:
return decimal.Decimal(money.amount)
except:
return decimal.Decimal(0)
def refilling_to_pb(refilling: models.Refilling) -> com_pb2.Refilling:
return com_pb2.Refilling(
id=refilling.id,
customer_id=refilling.customer_id,
counter_id=refilling.counter_id,
device_uuid=refilling.machine_id,
payment_method=refilling.payment_method.id,
amount=decimal_to_pb_money(refilling.amount),
cancelled=refilling.cancelled,
date=date_to_pb(refilling.date),
)
from . import db, models, com_pb2, com_pb2_grpc
from .pbutils import (
pb_now,
date_to_pb,
decimal_to_pb_money,
pb_money_to_decimal,
refilling_to_pb,
)
def get_or_create_machine(_uuid: str) -> models.Machine:
......@@ -569,7 +541,7 @@ def serve(address: str, port: int, reflect: bool):
com_pb2_grpc.add_PaymentProtocolServicer_to_server(PaymentServicer(), server)
server.add_insecure_port("%s:%d" % (address, port))
print("Listening from %s:%d" % (address, port))
logging.info("Listening from %s:%d" % (address, port))
if reflect:
from grpc_reflection.v1alpha import reflection
......@@ -578,7 +550,7 @@ def serve(address: str, port: int, reflect: bool):
reflection.SERVICE_NAME,
)
reflection.enable_server_reflection(SERVICE_NAMES, server)
print("Reflection enabled")
logging.info("Reflection enabled")
server.start()
server.wait_for_termination()
# -*- coding:utf-8 -*
import logging
import os
DB_PATH = "database.sqlite"
try:
with open(os.environ.get("CFG", "settings_custom.py"), "r") as f:
exec(f.read())
logging.info("Custom settings imported")
except Exception as e:
logging.warning(f"Custom settings failed: {e}")
#!/usr/bin/env python3
# -*- coding:utf-8 -*
import unittest
import grpc
from google.protobuf.timestamp_pb2 import Timestamp
from grpc_testing import server_from_dictionary, strict_real_time
from . import db, models, com_pb2, com_pb2_grpc
from .pbutils import (
pb_now,
date_to_pb,
decimal_to_pb_money,
pb_money_to_decimal,
refilling_to_pb,
)
class PaymentProtocolTestCase(unittest.TestCase):
def __init__(self, method_name):
super().__init__(method_name)
servicers = {
com_pb2.DESCRIPTOR.services_by_name[
"PaymentProtocol"
]: com_pb2_grpc.PaymentProtocolServicer()
}
self.test_server = server_from_dictionary(servicers, strict_real_time())
def setUp(self):
# Create database
pass
class TestBalance(PaymentProtocolTestCase):
def test_non_existing_user(self):
print("HELLO")
self.assertEqual("hello", "hello")
def test_existing_user(self):
print("HELLO2")
self.assertEqual("hello", "hello")
from subprocess import Popen
import sys
import fileinput
def build_protoc():
p = Popen(
"%s -m grpc_tools.protoc -I%s --python_out=%s --grpc_python_out=%s %s"
% (sys.executable, "../protos", "./cashless_server/", "./cashless_server/", "../protos/com.proto"),
shell=True,
)
p.wait()
# Fix import path issues
with fileinput.FileInput("cashless_server/com_pb2_grpc.py", inplace=True) as file:
for line in file:
print(
line.replace("import com_pb2", "import cashless_server.com_pb2"), end="",
)
if __name__ == "__main__":
build_protoc()
\ No newline at end of file
[tool.poetry]
name = "cashless_server"
version = "0.1.0"
description = "Server for centralizing cashless system"
authors = ["Sli <antoine@bartuccio.fr>"]
license = "GPL-3.0-only"
include = [
{path = "cashless_server/com.proto"},
{path = "cashless_server/com_pb2.py"},
{path = "cashless_server/com_pb2_grpc.py"},
{path = "cashless_server/db_import_schema.json"}
]
[tool.poetry.build]
script = "build.py"
# generate-setup-file = false
[tool.poetry.scripts]
cash = 'cashless_server.manage:main'
[tool.poetry.dependencies]
python = "^3.8"
grpcio = "^1.41.0"
grpcio-tools = "^1.41.0"
grpcio-reflection = "^1.41.0"
click = "^8.0.3"
jsonschema = "^4.1.2"
python-slugify = "^5.0.2"
SQLAlchemy = "^1.4.26"
grpcio-testing = {version = "^1.41.0", optional= true }
pytest = "^6.2.5"
[tool.poetry.extras]
testing = ["grpcio-testing"]
[tool.poetry.dev-dependencies]
[build-system]
requires = ["poetry-core>=1.0.0", "wheel", "setuptools", "grpcio-tools", "click"]
build-backend = "poetry.core.masonry.api"
grpcio==1.36
grpcio-tools==1.36
grpcio-reflection==1.36
SQLAlchemy==1.3.17
click==7.1.2
jsonschema
python-slugify
# -*- coding:utf-8 -*
import sys
DB_PATH = "database.sqlite"
try:
from .settings_custom import *
print("Custom settings imported", file=sys.stderr)
except:
print("Custom settings failed", file=sys.stderr)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment