alexander/urban-meal-delivery
1
1
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity Actions

Add an ORM layer

Browse source 
- use SQLAlchemy (and PostgreSQL) to model the ORM layer
- add the following models:
  + Address => modelling all kinds of addresses
  + City => model the three target cities
  + Courier => model the UDP's couriers
  + Customer => model the UDP's customers
  + Order => model the orders received by the UDP
  + Restaurant => model the restaurants active on the UDP
- so far, the emphasis lies on expression the Foreign Key
  and Check Constraints that are used to validate the assumptions
  inherent to the cleanded data
- provide database-independent unit tests with 100% coverage
- provide additional integration tests ("e2e") that commit data to
  a PostgreSQL instance to validate that the constraints work
- adapt linting rules a bit
This commit is contained in:
Alexander Hess 2020-08-09 03:45:19 +02:00
parent d219fa816d
commit fdcc93a1ea
Signed by: alexander
GPG key ID: 344EA5AB10D868E0
24 changed files with 2119 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
noxfile.py
View file

@ -249,6 +249,8 @@ def test(session):
'--cov-branch',
'--cov-fail-under=100',
'--cov-report=term-missing:skip-covered',
'-k',
'not e2e',
PYTEST_LOCATION,
)
session.run('pytest', '--version')

75
poetry.lock generated
View file

@ -700,6 +700,14 @@ pyyaml = ">=5.1"
toml = "*"
virtualenv = ">=20.0.8"
[[package]]
category = "main"
description = "psycopg2 - Python-PostgreSQL Database Adapter"
name = "psycopg2"
optional = false
python-versions = ">=2.7,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*"
version = "2.8.5"
[[package]]
category = "dev"
description = "library with cross-python path, ini-parsing, io, code, log facilities"
@ -1010,6 +1018,26 @@ version = "1.1.4"
lint = ["flake8", "mypy", "docutils-stubs"]
test = ["pytest"]
[[package]]
category = "main"
description = "Database Abstraction Library"
name = "sqlalchemy"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
version = "1.3.18"
[package.extras]
mssql = ["pyodbc"]
mssql_pymssql = ["pymssql"]
mssql_pyodbc = ["pyodbc"]
mysql = ["mysqlclient"]
oracle = ["cx-oracle"]
postgresql = ["psycopg2"]
postgresql_pg8000 = ["pg8000"]
postgresql_psycopg2binary = ["psycopg2-binary"]
postgresql_psycopg2cffi = ["psycopg2cffi"]
pymysql = ["pymysql"]
[[package]]
category = "dev"
description = "Manage dynamic plugins for Python applications"
@ -1151,7 +1179,7 @@ optional = ["pygments", "colorama"]
tests = ["pytest", "pytest-cov", "codecov", "scikit-build", "cmake", "ninja", "pybind11"]
[metadata]
content-hash = "862a0bc650dab485af541f185ed3c1e86a8947e7518c8fbcacfd19d5b8055af5"
content-hash = "508cbaa3105e47cac64c68663ed8d4178ee752bf267cb24cf68264e73325e10b"
lock-version = "1.0"
python-versions = "^3.8"
@ -1501,6 +1529,21 @@ pre-commit = [
{file = "pre_commit-2.6.0-py2.py3-none-any.whl", hash = "sha256:e8b1315c585052e729ab7e99dcca5698266bedce9067d21dc909c23e3ceed626"},
{file = "pre_commit-2.6.0.tar.gz", hash = "sha256:1657663fdd63a321a4a739915d7d03baedd555b25054449090f97bb0cb30a915"},
]
psycopg2 = [
{file = "psycopg2-2.8.5-cp27-cp27m-win32.whl", hash = "sha256:a0984ff49e176062fcdc8a5a2a670c9bb1704a2f69548bce8f8a7bad41c661bf"},
{file = "psycopg2-2.8.5-cp27-cp27m-win_amd64.whl", hash = "sha256:acf56d564e443e3dea152efe972b1434058244298a94348fc518d6dd6a9fb0bb"},
{file = "psycopg2-2.8.5-cp34-cp34m-win32.whl", hash = "sha256:440a3ea2c955e89321a138eb7582aa1d22fe286c7d65e26a2c5411af0a88ae72"},
{file = "psycopg2-2.8.5-cp34-cp34m-win_amd64.whl", hash = "sha256:6b306dae53ec7f4f67a10942cf8ac85de930ea90e9903e2df4001f69b7833f7e"},
{file = "psycopg2-2.8.5-cp35-cp35m-win32.whl", hash = "sha256:d3b29d717d39d3580efd760a9a46a7418408acebbb784717c90d708c9ed5f055"},
{file = "psycopg2-2.8.5-cp35-cp35m-win_amd64.whl", hash = "sha256:6a471d4d2a6f14c97a882e8d3124869bc623f3df6177eefe02994ea41fd45b52"},
{file = "psycopg2-2.8.5-cp36-cp36m-win32.whl", hash = "sha256:27c633f2d5db0fc27b51f1b08f410715b59fa3802987aec91aeb8f562724e95c"},
{file = "psycopg2-2.8.5-cp36-cp36m-win_amd64.whl", hash = "sha256:2df2bf1b87305bd95eb3ac666ee1f00a9c83d10927b8144e8e39644218f4cf81"},
{file = "psycopg2-2.8.5-cp37-cp37m-win32.whl", hash = "sha256:ac5b23d0199c012ad91ed1bbb971b7666da651c6371529b1be8cbe2a7bf3c3a9"},
{file = "psycopg2-2.8.5-cp37-cp37m-win_amd64.whl", hash = "sha256:2c0afb40cfb4d53487ee2ebe128649028c9a78d2476d14a67781e45dc287f080"},
{file = "psycopg2-2.8.5-cp38-cp38-win32.whl", hash = "sha256:2327bf42c1744a434ed8ed0bbaa9168cac7ee5a22a9001f6fc85c33b8a4a14b7"},
{file = "psycopg2-2.8.5-cp38-cp38-win_amd64.whl", hash = "sha256:132efc7ee46a763e68a815f4d26223d9c679953cd190f1f218187cb60decf535"},
{file = "psycopg2-2.8.5.tar.gz", hash = "sha256:f7d46240f7a1ae1dd95aab38bd74f7428d46531f69219954266d669da60c0818"},
]
py = [
{file = "py-1.9.0-py2.py3-none-any.whl", hash = "sha256:366389d1db726cd2fcfc79732e75410e5fe4d31db13692115529d34069a043c2"},
{file = "py-1.9.0.tar.gz", hash = "sha256:9ca6883ce56b4e8da7e79ac18787889fa5206c79dcc67fb065376cd2fe03f342"},
@ -1632,6 +1675,36 @@ sphinxcontrib-serializinghtml = [
{file = "sphinxcontrib-serializinghtml-1.1.4.tar.gz", hash = "sha256:eaa0eccc86e982a9b939b2b82d12cc5d013385ba5eadcc7e4fed23f4405f77bc"},
{file = "sphinxcontrib_serializinghtml-1.1.4-py2.py3-none-any.whl", hash = "sha256:f242a81d423f59617a8e5cf16f5d4d74e28ee9a66f9e5b637a18082991db5a9a"},
]
sqlalchemy = [
{file = "SQLAlchemy-1.3.18-cp27-cp27m-macosx_10_14_x86_64.whl", hash = "sha256:f11c2437fb5f812d020932119ba02d9e2bc29a6eca01a055233a8b449e3e1e7d"},
{file = "SQLAlchemy-1.3.18-cp27-cp27m-manylinux1_x86_64.whl", hash = "sha256:0ec575db1b54909750332c2e335c2bb11257883914a03bc5a3306a4488ecc772"},
{file = "SQLAlchemy-1.3.18-cp27-cp27m-manylinux2010_x86_64.whl", hash = "sha256:f57be5673e12763dd400fea568608700a63ce1c6bd5bdbc3cc3a2c5fdb045274"},
{file = "SQLAlchemy-1.3.18-cp27-cp27m-win32.whl", hash = "sha256:8cac7bb373a5f1423e28de3fd5fc8063b9c8ffe8957dc1b1a59cb90453db6da1"},
{file = "SQLAlchemy-1.3.18-cp27-cp27m-win_amd64.whl", hash = "sha256:adad60eea2c4c2a1875eb6305a0b6e61a83163f8e233586a4d6a55221ef984fe"},
{file = "SQLAlchemy-1.3.18-cp27-cp27mu-manylinux1_x86_64.whl", hash = "sha256:57aa843b783179ab72e863512e14bdcba186641daf69e4e3a5761d705dcc35b1"},
{file = "SQLAlchemy-1.3.18-cp27-cp27mu-manylinux2010_x86_64.whl", hash = "sha256:621f58cd921cd71ba6215c42954ffaa8a918eecd8c535d97befa1a8acad986dd"},
{file = "SQLAlchemy-1.3.18-cp35-cp35m-macosx_10_14_x86_64.whl", hash = "sha256:fc728ece3d5c772c196fd338a99798e7efac7a04f9cb6416299a3638ee9a94cd"},
{file = "SQLAlchemy-1.3.18-cp35-cp35m-manylinux1_x86_64.whl", hash = "sha256:736d41cfebedecc6f159fc4ac0769dc89528a989471dc1d378ba07d29a60ba1c"},
{file = "SQLAlchemy-1.3.18-cp35-cp35m-manylinux2010_x86_64.whl", hash = "sha256:427273b08efc16a85aa2b39892817e78e3ed074fcb89b2a51c4979bae7e7ba98"},
{file = "SQLAlchemy-1.3.18-cp35-cp35m-win32.whl", hash = "sha256:cbe1324ef52ff26ccde2cb84b8593c8bf930069dfc06c1e616f1bfd4e47f48a3"},
{file = "SQLAlchemy-1.3.18-cp35-cp35m-win_amd64.whl", hash = "sha256:8fd452dc3d49b3cc54483e033de6c006c304432e6f84b74d7b2c68afa2569ae5"},
{file = "SQLAlchemy-1.3.18-cp36-cp36m-macosx_10_14_x86_64.whl", hash = "sha256:e89e0d9e106f8a9180a4ca92a6adde60c58b1b0299e1b43bd5e0312f535fbf33"},
{file = "SQLAlchemy-1.3.18-cp36-cp36m-manylinux1_x86_64.whl", hash = "sha256:6ac2558631a81b85e7fb7a44e5035347938b0a73f5fdc27a8566777d0792a6a4"},
{file = "SQLAlchemy-1.3.18-cp36-cp36m-manylinux2010_x86_64.whl", hash = "sha256:87fad64529cde4f1914a5b9c383628e1a8f9e3930304c09cf22c2ae118a1280e"},
{file = "SQLAlchemy-1.3.18-cp36-cp36m-win32.whl", hash = "sha256:e4624d7edb2576cd72bb83636cd71c8ce544d8e272f308bd80885056972ca299"},
{file = "SQLAlchemy-1.3.18-cp36-cp36m-win_amd64.whl", hash = "sha256:89494df7f93b1836cae210c42864b292f9b31eeabca4810193761990dc689cce"},
{file = "SQLAlchemy-1.3.18-cp37-cp37m-macosx_10_14_x86_64.whl", hash = "sha256:716754d0b5490bdcf68e1e4925edc02ac07209883314ad01a137642ddb2056f1"},
{file = "SQLAlchemy-1.3.18-cp37-cp37m-manylinux1_x86_64.whl", hash = "sha256:50c4ee32f0e1581828843267d8de35c3298e86ceecd5e9017dc45788be70a864"},
{file = "SQLAlchemy-1.3.18-cp37-cp37m-manylinux2010_x86_64.whl", hash = "sha256:d98bc827a1293ae767c8f2f18be3bb5151fd37ddcd7da2a5f9581baeeb7a3fa1"},
{file = "SQLAlchemy-1.3.18-cp37-cp37m-win32.whl", hash = "sha256:0942a3a0df3f6131580eddd26d99071b48cfe5aaf3eab2783076fbc5a1c1882e"},
{file = "SQLAlchemy-1.3.18-cp37-cp37m-win_amd64.whl", hash = "sha256:16593fd748944726540cd20f7e83afec816c2ac96b082e26ae226e8f7e9688cf"},
{file = "SQLAlchemy-1.3.18-cp38-cp38-macosx_10_14_x86_64.whl", hash = "sha256:c26f95e7609b821b5f08a72dab929baa0d685406b953efd7c89423a511d5c413"},
{file = "SQLAlchemy-1.3.18-cp38-cp38-manylinux1_x86_64.whl", hash = "sha256:512a85c3c8c3995cc91af3e90f38f460da5d3cade8dc3a229c8e0879037547c9"},
{file = "SQLAlchemy-1.3.18-cp38-cp38-manylinux2010_x86_64.whl", hash = "sha256:d05c4adae06bd0c7f696ae3ec8d993ed8ffcc4e11a76b1b35a5af8a099bd2284"},
{file = "SQLAlchemy-1.3.18-cp38-cp38-win32.whl", hash = "sha256:109581ccc8915001e8037b73c29590e78ce74be49ca0a3630a23831f9e3ed6c7"},
{file = "SQLAlchemy-1.3.18-cp38-cp38-win_amd64.whl", hash = "sha256:8619b86cb68b185a778635be5b3e6018623c0761dde4df2f112896424aa27bd8"},
{file = "SQLAlchemy-1.3.18.tar.gz", hash = "sha256:da2fb75f64792c1fc64c82313a00c728a7c301efe6a60b7a9fe35b16b4368ce7"},
]
stevedore = [
{file = "stevedore-3.2.0-py3-none-any.whl", hash = "sha256:c8f4f0ebbc394e52ddf49de8bcc3cf8ad2b4425ebac494106bbc5e3661ac7633"},
{file = "stevedore-3.2.0.tar.gz", hash = "sha256:38791aa5bed922b0a844513c5f9ed37774b68edc609e5ab8ab8d8fe0ce4315e5"},

2
pyproject.toml
View file

@ -28,7 +28,9 @@ repository = "https://github.com/webartifex/urban-meal-delivery"
python = "^3.8"
click = "^7.1.2"
psycopg2 = "^2.8.5" # adapter for PostgreSQL
python-dotenv = "^0.14.0"
sqlalchemy = "^1.3.18"
[tool.poetry.dev-dependencies]
# Task Runners

17
setup.cfg
View file

@ -84,6 +84,8 @@ ignore =
# If --ignore is passed on the command
# line, still ignore the following:
extend-ignore =
# Too long line => duplicate with E501.
B950,
# Comply with black's style.
# Source: https://github.com/psf/black/blob/master/docs/compatible_configs.md#flake8
E203, W503,
@ -114,6 +116,10 @@ per-file-ignores =
WPS115,
# Numbers are normal in config files.
WPS432,
src/urban_meal_delivery/db/addresses.py:
WPS226,
src/urban_meal_delivery/db/orders.py:
WPS226,
tests/*.py:
# Type annotations are not strictly enforced.
ANN0, ANN2,
@ -121,8 +127,12 @@ per-file-ignores =
S101,
# Shadowing outer scopes occurs naturally with mocks.
WPS442,
# Modules may have many test cases.
WPS202,WPS204,WPS214,
# No overuse of string constants (e.g., '__version__').
WPS226,
# Numbers are normal in test cases as expected results.
WPS432,
# Explicitly set mccabe's maximum complexity to 10 as recommended by
# Thomas McCabe, the inventor of the McCabe complexity, and the NIST.
@ -199,6 +209,8 @@ ignore_missing_imports = true
ignore_missing_imports = true
[mypy-pytest]
ignore_missing_imports = true
[mypy-sqlalchemy.*]
ignore_missing_imports = true
[pylint.FORMAT]
@ -210,6 +222,9 @@ disable =
# We use TODO's to indicate locations in the source base
# that must be worked on in the near future.
fixme,
# Too many false positives and cannot be disabled within a file.
# Source: https://github.com/PyCQA/pylint/issues/214
duplicate-code,
# Comply with black's style.
bad-continuation, bad-whitespace,
# =====================
@ -240,3 +255,5 @@ addopts =
--strict-markers
cache_dir = .cache/pytest
console_output_style = count
markers =
e2e: integration tests, inlc., for example, tests touching a database

9
src/urban_meal_delivery/__init__.py
View file

@ -27,4 +27,11 @@ else:
# Little Hack: "Overwrites" the config module so that the environment is already set.
config = _config.get_config('testing' if _os.getenv('TESTING') else 'production')
config: _config.Config = _config.get_config(
'testing' if _os.getenv('TESTING') else 'production',
)
# Import `db` down here as it depends on `config`.
# pylint:disable=wrong-import-position
from urban_meal_delivery import db # noqa:E402,F401 isort:skip

12
src/urban_meal_delivery/_config.py
View file

@ -6,9 +6,10 @@ via the `config` proxy at the package's top level.
That already loads the correct configuration
depending on the current environment.
"""
import datetime
import os
import random
import string
import warnings
import dotenv
@ -17,6 +18,13 @@ import dotenv
dotenv.load_dotenv()
def random_schema_name() -> str:
"""Generate a random PostgreSQL schema name for testing."""
return ''.join(
random.choice(string.ascii_lowercase) for _ in range(10) # noqa:S311
)
class Config:
"""Configuration that applies in all situations."""
@ -57,7 +65,7 @@ class TestingConfig(Config):
TESTING = True
DATABASE_URI = os.getenv('DATABASE_URI_TESTING') or Config.DATABASE_URI
CLEAN_SCHEMA = os.getenv('CLEAN_SCHEMA_TESTING') or Config.CLEAN_SCHEMA
CLEAN_SCHEMA = os.getenv('CLEAN_SCHEMA_TESTING') or random_schema_name()
def get_config(env: str = 'production') -> Config:

11
src/urban_meal_delivery/db/__init__.py Normal file
View file

@ -0,0 +1,11 @@
"""Provide the ORM models and a connection to the database."""
from urban_meal_delivery.db.addresses import Address # noqa:F401
from urban_meal_delivery.db.cities import City # noqa:F401
from urban_meal_delivery.db.connection import make_engine # noqa:F401
from urban_meal_delivery.db.connection import make_session_factory # noqa:F401
from urban_meal_delivery.db.couriers import Courier # noqa:F401
from urban_meal_delivery.db.customers import Customer # noqa:F401
from urban_meal_delivery.db.meta import Base # noqa:F401
from urban_meal_delivery.db.orders import Order # noqa:F401
from urban_meal_delivery.db.restaurants import Restaurant # noqa:F401

82
src/urban_meal_delivery/db/addresses.py Normal file
View file

@ -0,0 +1,82 @@
"""Provide the ORM's Address model."""
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext import hybrid
from urban_meal_delivery.db import meta
class Address(meta.Base):
"""An Address of a Customer or a Restaurant on the UDP."""
__tablename__ = 'addresses'
# Columns
id = sa.Column(sa.Integer, primary_key=True, autoincrement=False) # noqa:WPS125
_primary_id = sa.Column('primary_id', sa.Integer, nullable=False, index=True)
created_at = sa.Column(sa.DateTime, nullable=False)
place_id = sa.Column(
sa.Unicode(length=120), nullable=False, index=True, # noqa:WPS432
)
latitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
longitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
_city_id = sa.Column('city_id', sa.SmallInteger, nullable=False, index=True)
city_name = sa.Column('city', sa.Unicode(length=25), nullable=False) # noqa:WPS432
zip_code = sa.Column(sa.Integer, nullable=False, index=True)
street = sa.Column(sa.Unicode(length=80), nullable=False) # noqa:WPS432
floor = sa.Column(sa.SmallInteger)
# Constraints
__table_args__ = (
sa.ForeignKeyConstraint(
['primary_id'], ['addresses.id'], onupdate='RESTRICT', ondelete='RESTRICT',
),
sa.ForeignKeyConstraint(
['city_id'], ['cities.id'], onupdate='RESTRICT', ondelete='RESTRICT',
),
sa.CheckConstraint(
'-90 <= latitude AND latitude <= 90', name='latitude_between_90_degrees',
),
sa.CheckConstraint(
'-180 <= longitude AND longitude <= 180',
name='longitude_between_180_degrees',
),
sa.CheckConstraint(
'30000 <= zip_code AND zip_code <= 99999', name='valid_zip_code',
),
sa.CheckConstraint('0 <= floor AND floor <= 40', name='realistic_floor'),
)
# Relationships
city = orm.relationship('City', back_populates='addresses')
restaurant = orm.relationship('Restaurant', back_populates='address', uselist=False)
orders_picked_up = orm.relationship(
'Order',
back_populates='pickup_address',
foreign_keys='[Order._pickup_address_id]',
)
orders_delivered = orm.relationship(
'Order',
back_populates='delivery_address',
foreign_keys='[Order._delivery_address_id]',
)
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}({street} in {city})>'.format(
cls=self.__class__.__name__, street=self.street, city=self.city_name,
)
@hybrid.hybrid_property
def is_primary(self) -> bool:
"""If an Address object is the earliest one entered at its location.
Street addresses may have been entered several times with different
versions/spellings of the street name and/or different floors.
`is_primary` indicates the first in a group of addresses.
"""
return self.id == self._primary_id

83
src/urban_meal_delivery/db/cities.py Normal file
View file

@ -0,0 +1,83 @@
"""Provide the ORM's City model."""
from typing import Dict
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.dialects import postgresql
from urban_meal_delivery.db import meta
class City(meta.Base):
"""A City where the UDP operates in."""
__tablename__ = 'cities'
# Generic columns
id = sa.Column( # noqa:WPS125
sa.SmallInteger, primary_key=True, autoincrement=False,
)
name = sa.Column(sa.Unicode(length=10), nullable=False)
kml = sa.Column(sa.UnicodeText, nullable=False)
# Google Maps related columns
_center_latitude = sa.Column(
'center_latitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
_center_longitude = sa.Column(
'center_longitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
_northeast_latitude = sa.Column(
'northeast_latitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
_northeast_longitude = sa.Column(
'northeast_longitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
_southwest_latitude = sa.Column(
'southwest_latitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
_southwest_longitude = sa.Column(
'southwest_longitude', postgresql.DOUBLE_PRECISION, nullable=False,
)
initial_zoom = sa.Column(sa.SmallInteger, nullable=False)
# Relationships
addresses = orm.relationship('Address', back_populates='city')
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}({name})>'.format(cls=self.__class__.__name__, name=self.name)
@property
def location(self) -> Dict[str, float]:
"""GPS location of the city's center.
Example:
{"latitude": 48.856614, "longitude": 2.3522219}
"""
return {
'latitude': self._center_latitude,
'longitude': self._center_longitude,
}
@property
def viewport(self) -> Dict[str, Dict[str, float]]:
"""Google Maps viewport of the city.
Example:
{
'northeast': {'latitude': 48.9021449, 'longitude': 2.4699208},
'southwest': {'latitude': 48.815573, 'longitude': 2.225193},
}
""" # noqa:RST203
return {
'northeast': {
'latitude': self._northeast_latitude,
'longitude': self._northeast_longitude,
},
'southwest': {
'latitude': self._southwest_latitude,
'longitude': self._southwest_longitude,
},
}

17
src/urban_meal_delivery/db/connection.py Normal file
View file

@ -0,0 +1,17 @@
"""Provide connection utils for the ORM layer."""
import sqlalchemy as sa
from sqlalchemy import engine
from sqlalchemy import orm
import urban_meal_delivery
def make_engine() -> engine.Engine: # pragma: no cover
"""Provide a configured Engine object."""
return sa.create_engine(urban_meal_delivery.config.DATABASE_URI)
def make_session_factory() -> orm.Session: # pragma: no cover
"""Provide a configured Session factory."""
return orm.sessionmaker(bind=make_engine())

51
src/urban_meal_delivery/db/couriers.py Normal file
View file

@ -0,0 +1,51 @@
"""Provide the ORM's Courier model."""
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.dialects import postgresql
from urban_meal_delivery.db import meta
class Courier(meta.Base):
"""A Courier working for the UDP."""
# pylint:disable=too-few-public-methods
__tablename__ = 'couriers'
# Columns
id = sa.Column(sa.Integer, primary_key=True, autoincrement=False) # noqa:WPS125
created_at = sa.Column(sa.DateTime, nullable=False)
vehicle = sa.Column(sa.Unicode(length=10), nullable=False)
historic_speed = sa.Column('speed', postgresql.DOUBLE_PRECISION, nullable=False)
capacity = sa.Column(sa.SmallInteger, nullable=False)
pay_per_hour = sa.Column(sa.SmallInteger, nullable=False)
pay_per_order = sa.Column(sa.SmallInteger, nullable=False)
# Constraints
__table_args__ = (
sa.CheckConstraint(
"vehicle IN ('bicycle', 'motorcycle')", name='available_vehicle_types',
),
sa.CheckConstraint('0 <= speed AND speed <= 30', name='realistic_speed'),
sa.CheckConstraint(
'0 <= capacity AND capacity <= 200', name='capacity_under_200_liters',
),
sa.CheckConstraint(
'0 <= pay_per_hour AND pay_per_hour <= 1500', name='realistic_pay_per_hour',
),
sa.CheckConstraint(
'0 <= pay_per_order AND pay_per_order <= 650',
name='realistic_pay_per_order',
),
)
# Relationships
orders = orm.relationship('Order', back_populates='courier')
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}(#{courier_id})>'.format(
cls=self.__class__.__name__, courier_id=self.id,
)

26
src/urban_meal_delivery/db/customers.py Normal file
View file

@ -0,0 +1,26 @@
"""Provide the ORM's Customer model."""
import sqlalchemy as sa
from sqlalchemy import orm
from urban_meal_delivery.db import meta
class Customer(meta.Base):
"""A Customer of the UDP."""
# pylint:disable=too-few-public-methods
__tablename__ = 'customers'
# Columns
id = sa.Column(sa.Integer, primary_key=True, autoincrement=False) # noqa:WPS125
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}(#{customer_id})>'.format(
cls=self.__class__.__name__, customer_id=self.id,
)
# Relationships
orders = orm.relationship('Order', back_populates='customer')

22
src/urban_meal_delivery/db/meta.py Normal file
View file

@ -0,0 +1,22 @@
"""Provide the ORM's declarative base."""
from typing import Any
import sqlalchemy as sa
from sqlalchemy.ext import declarative
import urban_meal_delivery
Base: Any = declarative.declarative_base(
metadata=sa.MetaData(
schema=urban_meal_delivery.config.CLEAN_SCHEMA,
naming_convention={
'pk': 'pk_%(table_name)s', # noqa:WPS323
'fk': 'fk_%(table_name)s_to_%(referred_table_name)s_via_%(column_0_N_name)s', # noqa:E501,WPS323
'uq': 'uq_%(table_name)s_on_%(column_0_N_name)s', # noqa:WPS323
'ix': 'ix_%(table_name)s_on_%(column_0_N_name)s', # noqa:WPS323
'ck': 'ck_%(table_name)s_on_%(constraint_name)s', # noqa:WPS323
},
),
)

526
src/urban_meal_delivery/db/orders.py Normal file
View file

@ -0,0 +1,526 @@
"""Provide the ORM's Order model."""
import datetime
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.dialects import postgresql
from urban_meal_delivery.db import meta
class Order(meta.Base): # noqa:WPS214
"""An Order by a Customer of the UDP."""
__tablename__ = 'orders'
# Generic columns
id = sa.Column(sa.Integer, primary_key=True, autoincrement=False) # noqa:WPS125
_delivery_id = sa.Column('delivery_id', sa.Integer, index=True, unique=True)
_customer_id = sa.Column('customer_id', sa.Integer, nullable=False, index=True)
placed_at = sa.Column(sa.DateTime, nullable=False, index=True)
ad_hoc = sa.Column(sa.Boolean, nullable=False)
scheduled_delivery_at = sa.Column(sa.DateTime, index=True)
scheduled_delivery_at_corrected = sa.Column(sa.Boolean, index=True)
first_estimated_delivery_at = sa.Column(sa.DateTime)
cancelled = sa.Column(sa.Boolean, nullable=False, index=True)
cancelled_at = sa.Column(sa.DateTime)
cancelled_at_corrected = sa.Column(sa.Boolean, index=True)
# Price-related columns
sub_total = sa.Column(sa.Integer, nullable=False)
delivery_fee = sa.Column(sa.SmallInteger, nullable=False)
total = sa.Column(sa.Integer, nullable=False)
# Restaurant-related columns
_restaurant_id = sa.Column(
'restaurant_id', sa.SmallInteger, nullable=False, index=True,
)
restaurant_notified_at = sa.Column(sa.DateTime)
restaurant_notified_at_corrected = sa.Column(sa.Boolean, index=True)
restaurant_confirmed_at = sa.Column(sa.DateTime)
restaurant_confirmed_at_corrected = sa.Column(sa.Boolean, index=True)
estimated_prep_duration = sa.Column(sa.Integer, index=True)
estimated_prep_duration_corrected = sa.Column(sa.Boolean, index=True)
estimated_prep_buffer = sa.Column(sa.Integer, nullable=False, index=True)
# Dispatch-related columns
_courier_id = sa.Column('courier_id', sa.Integer, index=True)
dispatch_at = sa.Column(sa.DateTime)
dispatch_at_corrected = sa.Column(sa.Boolean, index=True)
courier_notified_at = sa.Column(sa.DateTime)
courier_notified_at_corrected = sa.Column(sa.Boolean, index=True)
courier_accepted_at = sa.Column(sa.DateTime)
courier_accepted_at_corrected = sa.Column(sa.Boolean, index=True)
utilization = sa.Column(sa.SmallInteger, nullable=False)
# Pickup-related columns
_pickup_address_id = sa.Column(
'pickup_address_id', sa.Integer, nullable=False, index=True,
)
reached_pickup_at = sa.Column(sa.DateTime)
pickup_at = sa.Column(sa.DateTime)
pickup_at_corrected = sa.Column(sa.Boolean, index=True)
pickup_not_confirmed = sa.Column(sa.Boolean)
left_pickup_at = sa.Column(sa.DateTime)
left_pickup_at_corrected = sa.Column(sa.Boolean, index=True)
# Delivery-related columns
_delivery_address_id = sa.Column(
'delivery_address_id', sa.Integer, nullable=False, index=True,
)
reached_delivery_at = sa.Column(sa.DateTime)
delivery_at = sa.Column(sa.DateTime)
delivery_at_corrected = sa.Column(sa.Boolean, index=True)
delivery_not_confirmed = sa.Column(sa.Boolean)
_courier_waited_at_delivery = sa.Column('courier_waited_at_delivery', sa.Boolean)
# Statistical columns
logged_delivery_distance = sa.Column(sa.SmallInteger, nullable=True)
logged_avg_speed = sa.Column(postgresql.DOUBLE_PRECISION, nullable=True)
logged_avg_speed_distance = sa.Column(sa.SmallInteger, nullable=True)
# Constraints
__table_args__ = (
sa.ForeignKeyConstraint(
['customer_id'], ['customers.id'], onupdate='RESTRICT', ondelete='RESTRICT',
),
sa.ForeignKeyConstraint(
['restaurant_id'],
['restaurants.id'],
onupdate='RESTRICT',
ondelete='RESTRICT',
),
sa.ForeignKeyConstraint(
['courier_id'], ['couriers.id'], onupdate='RESTRICT', ondelete='RESTRICT',
),
sa.ForeignKeyConstraint(
['pickup_address_id'],
['addresses.id'],
onupdate='RESTRICT',
ondelete='RESTRICT',
),
sa.ForeignKeyConstraint(
['delivery_address_id'],
['addresses.id'],
onupdate='RESTRICT',
ondelete='RESTRICT',
),
sa.CheckConstraint(
"""
(ad_hoc IS TRUE AND scheduled_delivery_at IS NULL)
OR
(ad_hoc IS FALSE AND scheduled_delivery_at IS NOT NULL)
""",
name='either_ad_hoc_or_scheduled_order',
),
sa.CheckConstraint(
"""
NOT (
ad_hoc IS TRUE
AND (
EXTRACT(HOUR FROM placed_at) < 11
OR
EXTRACT(HOUR FROM placed_at) > 22
)
)
""",
name='ad_hoc_orders_within_business_hours',
),
sa.CheckConstraint(
"""
NOT (
ad_hoc IS FALSE
AND (
(
EXTRACT(HOUR FROM scheduled_delivery_at) <= 11
AND
NOT (
EXTRACT(HOUR FROM scheduled_delivery_at) = 11
AND
EXTRACT(MINUTE FROM scheduled_delivery_at) = 45
)
)
OR
EXTRACT(HOUR FROM scheduled_delivery_at) > 22
)
)
""",
name='scheduled_orders_within_business_hours',
),
sa.CheckConstraint(
"""
NOT (
EXTRACT(EPOCH FROM scheduled_delivery_at - placed_at) < 1800
)
""",
name='scheduled_orders_not_within_30_minutes',
),
sa.CheckConstraint(
"""
NOT (
cancelled IS FALSE
AND
cancelled_at IS NOT NULL
)
""",
name='only_cancelled_orders_may_have_cancelled_at',
),
sa.CheckConstraint(
"""
NOT (
cancelled IS TRUE
AND
delivery_at IS NOT NULL
)
""",
name='cancelled_orders_must_not_be_delivered',
),
sa.CheckConstraint(
'0 <= estimated_prep_duration AND estimated_prep_duration <= 2700',
name='estimated_prep_duration_between_0_and_2700',
),
sa.CheckConstraint(
'estimated_prep_duration % 60 = 0',
name='estimated_prep_duration_must_be_whole_minutes',
),
sa.CheckConstraint(
'0 <= estimated_prep_buffer AND estimated_prep_buffer <= 900',
name='estimated_prep_buffer_between_0_and_900',
),
sa.CheckConstraint(
'estimated_prep_buffer % 60 = 0',
name='estimated_prep_buffer_must_be_whole_minutes',
),
sa.CheckConstraint(
'0 <= utilization AND utilization <= 100',
name='utilization_between_0_and_100',
),
*(
sa.CheckConstraint(
f"""
({column} IS NULL AND {column}_corrected IS NULL)
OR
({column} IS NULL AND {column}_corrected IS TRUE)
OR
({column} IS NOT NULL AND {column}_corrected IS NOT NULL)
""",
name=f'corrections_only_for_set_value_{index}',
)
for index, column in enumerate(
(
'scheduled_delivery_at',
'cancelled_at',
'restaurant_notified_at',
'restaurant_confirmed_at',
'estimated_prep_duration',
'dispatch_at',
'courier_notified_at',
'courier_accepted_at',
'pickup_at',
'left_pickup_at',
'delivery_at',
),
)
),
*(
sa.CheckConstraint(
f"""
({event}_at IS NULL AND {event}_not_confirmed IS NULL)
OR
({event}_at IS NOT NULL AND {event}_not_confirmed IS NOT NULL)
""",
name=f'{event}_not_confirmed_only_if_{event}',
)
for event in ('pickup', 'delivery')
),
sa.CheckConstraint(
"""
(delivery_at IS NULL AND courier_waited_at_delivery IS NULL)
OR
(delivery_at IS NOT NULL AND courier_waited_at_delivery IS NOT NULL)
""",
name='courier_waited_at_delivery_only_if_delivery',
),
*(
sa.CheckConstraint(
constraint, name='ordered_timestamps_{index}'.format(index=index),
)
for index, constraint in enumerate(
(
'placed_at < scheduled_delivery_at',
'placed_at < first_estimated_delivery_at',
'placed_at < cancelled_at',
'placed_at < restaurant_notified_at',
'placed_at < restaurant_confirmed_at',
'placed_at < dispatch_at',
'placed_at < courier_notified_at',
'placed_at < courier_accepted_at',
'placed_at < reached_pickup_at',
'placed_at < left_pickup_at',
'placed_at < reached_delivery_at',
'placed_at < delivery_at',
'cancelled_at > restaurant_notified_at',
'cancelled_at > restaurant_confirmed_at',
'cancelled_at > dispatch_at',
'cancelled_at > courier_notified_at',
'cancelled_at > courier_accepted_at',
'cancelled_at > reached_pickup_at',
'cancelled_at > pickup_at',
'cancelled_at > left_pickup_at',
'cancelled_at > reached_delivery_at',
'cancelled_at > delivery_at',
'restaurant_notified_at < restaurant_confirmed_at',
'restaurant_notified_at < pickup_at',
'restaurant_confirmed_at < pickup_at',
'dispatch_at < courier_notified_at',
'dispatch_at < courier_accepted_at',
'dispatch_at < reached_pickup_at',
'dispatch_at < pickup_at',
'dispatch_at < left_pickup_at',
'dispatch_at < reached_delivery_at',
'dispatch_at < delivery_at',
'courier_notified_at < courier_accepted_at',
'courier_notified_at < reached_pickup_at',
'courier_notified_at < pickup_at',
'courier_notified_at < left_pickup_at',
'courier_notified_at < reached_delivery_at',
'courier_notified_at < delivery_at',
'courier_accepted_at < reached_pickup_at',
'courier_accepted_at < pickup_at',
'courier_accepted_at < left_pickup_at',
'courier_accepted_at < reached_delivery_at',
'courier_accepted_at < delivery_at',
'reached_pickup_at < pickup_at',
'reached_pickup_at < left_pickup_at',
'reached_pickup_at < reached_delivery_at',
'reached_pickup_at < delivery_at',
'pickup_at < left_pickup_at',
'pickup_at < reached_delivery_at',
'pickup_at < delivery_at',
'left_pickup_at < reached_delivery_at',
'left_pickup_at < delivery_at',
'reached_delivery_at < delivery_at',
),
)
),
)
# Relationships
customer = orm.relationship('Customer', back_populates='orders')
restaurant = orm.relationship('Restaurant', back_populates='orders')
courier = orm.relationship('Courier', back_populates='orders')
pickup_address = orm.relationship(
'Address',
back_populates='orders_picked_up',
foreign_keys='[Order._pickup_address_id]',
)
delivery_address = orm.relationship(
'Address',
back_populates='orders_delivered',
foreign_keys='[Order._delivery_address_id]',
)
# Convenience properties
@property
def scheduled(self) -> bool:
"""Inverse of Order.ad_hoc."""
return not self.ad_hoc
@property
def completed(self) -> bool:
"""Inverse of Order.cancelled."""
return not self.cancelled
@property
def corrected(self) -> bool:
"""If any timestamp was corrected as compared to the original data."""
return (
self.scheduled_delivery_at_corrected # noqa:WPS222 => too much logic
or self.cancelled_at_corrected
or self.restaurant_notified_at_corrected
or self.restaurant_confirmed_at_corrected
or self.dispatch_at_corrected
or self.courier_notified_at_corrected
or self.courier_accepted_at_corrected
or self.pickup_at_corrected
or self.left_pickup_at_corrected
or self.delivery_at_corrected
)
# Timing-related properties
@property
def time_to_accept(self) -> datetime.timedelta:
"""Time until a courier accepted an order.
This adds the time it took the UDP to notify a courier.
"""
if not self.dispatch_at:
raise RuntimeError('dispatch_at is not set')
if not self.courier_accepted_at:
raise RuntimeError('courier_accepted_at is not set')
return self.courier_accepted_at - self.dispatch_at
@property
def time_to_react(self) -> datetime.timedelta:
"""Time a courier took to accept an order.
This time is a subset of Order.time_to_accept.
"""
if not self.courier_notified_at:
raise RuntimeError('courier_notified_at is not set')
if not self.courier_accepted_at:
raise RuntimeError('courier_accepted_at is not set')
return self.courier_accepted_at - self.courier_notified_at
@property
def time_to_pickup(self) -> datetime.timedelta:
"""Time from a courier's acceptance to arrival at the pickup location."""
if not self.courier_accepted_at:
raise RuntimeError('courier_accepted_at is not set')
if not self.reached_pickup_at:
raise RuntimeError('reached_pickup_at is not set')
return self.reached_pickup_at - self.courier_accepted_at
@property
def time_at_pickup(self) -> datetime.timedelta:
"""Time a courier stayed at the pickup location."""
if not self.reached_pickup_at:
raise RuntimeError('reached_pickup_at is not set')
if not self.pickup_at:
raise RuntimeError('pickup_at is not set')
return self.pickup_at - self.reached_pickup_at
@property
def scheduled_pickup_at(self) -> datetime.datetime:
"""Point in time at which the pickup was scheduled."""
if not self.restaurant_notified_at:
raise RuntimeError('restaurant_notified_at is not set')
if not self.estimated_prep_duration:
raise RuntimeError('estimated_prep_duration is not set')
delta = datetime.timedelta(seconds=self.estimated_prep_duration)
return self.restaurant_notified_at + delta
@property
def courier_early(self) -> datetime.timedelta:
"""Time by which a courier is early for pickup.
Measured relative to Order.scheduled_pickup_at.
0 if the courier is on time or late.
Goes together with Order.courier_late.
"""
return max(
datetime.timedelta(), self.scheduled_pickup_at - self.reached_pickup_at,
)
@property
def courier_late(self) -> datetime.timedelta:
"""Time by which a courier is late for pickup.
Measured relative to Order.scheduled_pickup_at.
0 if the courier is on time or early.
Goes together with Order.courier_early.
"""
return max(
datetime.timedelta(), self.reached_pickup_at - self.scheduled_pickup_at,
)
@property
def restaurant_early(self) -> datetime.timedelta:
"""Time by which a restaurant is early for pickup.
Measured relative to Order.scheduled_pickup_at.
0 if the restaurant is on time or late.
Goes together with Order.restaurant_late.
"""
return max(datetime.timedelta(), self.scheduled_pickup_at - self.pickup_at)
@property
def restaurant_late(self) -> datetime.timedelta:
"""Time by which a restaurant is late for pickup.
Measured relative to Order.scheduled_pickup_at.
0 if the restaurant is on time or early.
Goes together with Order.restaurant_early.
"""
return max(datetime.timedelta(), self.pickup_at - self.scheduled_pickup_at)
@property
def time_to_delivery(self) -> datetime.timedelta:
"""Time a courier took from pickup to delivery location."""
if not self.pickup_at:
raise RuntimeError('pickup_at is not set')
if not self.reached_delivery_at:
raise RuntimeError('reached_delivery_at is not set')
return self.reached_delivery_at - self.pickup_at
@property
def time_at_delivery(self) -> datetime.timedelta:
"""Time a courier stayed at the delivery location."""
if not self.reached_delivery_at:
raise RuntimeError('reached_delivery_at is not set')
if not self.delivery_at:
raise RuntimeError('delivery_at is not set')
return self.delivery_at - self.reached_delivery_at
@property
def courier_waited_at_delivery(self) -> datetime.timedelta:
"""Time a courier waited at the delivery location."""
if self._courier_waited_at_delivery:
return self.time_at_delivery
return datetime.timedelta()
@property
def delivery_early(self) -> datetime.timedelta:
"""Time by which a scheduled order was early.
Measured relative to Order.scheduled_delivery_at.
0 if the delivery is on time or late.
Goes together with Order.delivery_late.
"""
if not self.scheduled:
raise AttributeError('Makes sense only for scheduled orders')
return max(datetime.timedelta(), self.scheduled_delivery_at - self.delivery_at)
@property
def delivery_late(self) -> datetime.timedelta:
"""Time by which a scheduled order was late.
Measured relative to Order.scheduled_delivery_at.
0 if the delivery is on time or early.
Goes together with Order.delivery_early.
"""
if not self.scheduled:
raise AttributeError('Makes sense only for scheduled orders')
return max(datetime.timedelta(), self.delivery_at - self.scheduled_delivery_at)
@property
def total_time(self) -> datetime.timedelta:
"""Time from order placement to delivery for an ad-hoc order."""
if self.scheduled:
raise AttributeError('Scheduled orders have no total_time')
if self.cancelled:
raise RuntimeError('Cancelled orders have no total_time')
return self.delivery_at - self.placed_at
# Other Methods
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}(#{order_id})>'.format(
cls=self.__class__.__name__, order_id=self.id,
)

42
src/urban_meal_delivery/db/restaurants.py Normal file
View file

@ -0,0 +1,42 @@
"""Provide the ORM's Restaurant model."""
import sqlalchemy as sa
from sqlalchemy import orm
from urban_meal_delivery.db import meta
class Restaurant(meta.Base):
"""A Restaurant selling meals on the UDP."""
# pylint:disable=too-few-public-methods
__tablename__ = 'restaurants'
# Columns
id = sa.Column( # noqa:WPS125
sa.SmallInteger, primary_key=True, autoincrement=False,
)
created_at = sa.Column(sa.DateTime, nullable=False)
name = sa.Column(sa.Unicode(length=45), nullable=False) # noqa:WPS432
_address_id = sa.Column('address_id', sa.Integer, nullable=False, index=True)
estimated_prep_duration = sa.Column(sa.SmallInteger, nullable=False)
# Constraints
__table_args__ = (
sa.ForeignKeyConstraint(
['address_id'], ['addresses.id'], onupdate='RESTRICT', ondelete='RESTRICT',
),
sa.CheckConstraint(
'0 <= estimated_prep_duration AND estimated_prep_duration <= 2400',
name='realistic_estimated_prep_duration',
),
)
# Relationships
address = orm.relationship('Address', back_populates='restaurant')
orders = orm.relationship('Order', back_populates='restaurant')
def __repr__(self) -> str:
"""Non-literal text representation."""
return '<{cls}({name})>'.format(cls=self.__class__.__name__, name=self.name)

1
tests/db/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Test the ORM layer."""

244
tests/db/conftest.py Normal file
View file

@ -0,0 +1,244 @@
"""Utils for testing the ORM layer."""
import datetime
import pytest
from sqlalchemy import schema
from urban_meal_delivery import config
from urban_meal_delivery import db
@pytest.fixture(scope='session')
def db_engine():
"""Create all tables given the ORM models.
The tables are put into a distinct PostgreSQL schema
that is removed after all tests are over.
The engine used to do that is yielded.
"""
engine = db.make_engine()
engine.execute(schema.CreateSchema(config.CLEAN_SCHEMA))
db.Base.metadata.create_all(engine)
try:
yield engine
finally:
engine.execute(schema.DropSchema(config.CLEAN_SCHEMA, cascade=True))
@pytest.fixture
def db_session(db_engine):
"""A SQLAlchemy session that rolls back everything after a test case."""
connection = db_engine.connect()
# Begin the outer most transaction
# that is rolled back at the end of the test.
transaction = connection.begin()
# Create a session bound on the same connection as the transaction.
# Using any other session would not work.
Session = db.make_session_factory() # noqa:N806
session = Session(bind=connection)
try:
yield session
finally:
session.close()
transaction.rollback()
connection.close()
@pytest.fixture
def address_data():
"""The data for an Address object in Paris."""
return {
'id': 1,
'_primary_id': 1, # => "itself"
'created_at': datetime.datetime(2020, 1, 2, 3, 4, 5),
'place_id': 'ChIJxSr71vZt5kcRoFHY4caCCxw',
'latitude': 48.85313,
'longitude': 2.37461,
'_city_id': 1,
'city_name': 'St. German',
'zip_code': '75011',
'street': '42 Rue De Charonne',
'floor': None,
}
@pytest.fixture
def address(address_data, city):
"""An Address object."""
address = db.Address(**address_data)
address.city = city
return address
@pytest.fixture
def address2_data():
"""The data for an Address object in Paris."""
return {
'id': 2,
'_primary_id': 2, # => "itself"
'created_at': datetime.datetime(2020, 1, 2, 4, 5, 6),
'place_id': 'ChIJs-9a6QZy5kcRY8Wwk9Ywzl8',
'latitude': 48.852196,
'longitude': 2.373937,
'_city_id': 1,
'city_name': 'Paris',
'zip_code': '75011',
'street': 'Rue De Charonne 3',
'floor': 2,
}
@pytest.fixture
def address2(address2_data, city):
"""An Address object."""
address2 = db.Address(**address2_data)
address2.city = city
return address2
@pytest.fixture
def city_data():
"""The data for the City object modeling Paris."""
return {
'id': 1,
'name': 'Paris',
'kml': "<?xml version='1.0' encoding='UTF-8'?> ...",
'_center_latitude': 48.856614,
'_center_longitude': 2.3522219,
'_northeast_latitude': 48.9021449,
'_northeast_longitude': 2.4699208,
'_southwest_latitude': 48.815573,
'_southwest_longitude': 2.225193,
'initial_zoom': 12,
}
@pytest.fixture
def city(city_data):
"""A City object."""
return db.City(**city_data)
@pytest.fixture
def courier_data():
"""The data for a Courier object."""
return {
'id': 1,
'created_at': datetime.datetime(2020, 1, 2, 3, 4, 5),
'vehicle': 'bicycle',
'historic_speed': 7.89,
'capacity': 100,
'pay_per_hour': 750,
'pay_per_order': 200,
}
@pytest.fixture
def courier(courier_data):
"""A Courier object."""
return db.Courier(**courier_data)
@pytest.fixture
def customer_data():
"""The data for the Customer object."""
return {'id': 1}
@pytest.fixture
def customer(customer_data):
"""A Customer object."""
return db.Customer(**customer_data)
@pytest.fixture
def order_data():
"""The data for an ad-hoc Order object."""
return {
'id': 1,
'_delivery_id': 1,
'_customer_id': 1,
'placed_at': datetime.datetime(2020, 1, 2, 11, 55, 11),
'ad_hoc': True,
'scheduled_delivery_at': None,
'scheduled_delivery_at_corrected': None,
'first_estimated_delivery_at': datetime.datetime(2020, 1, 2, 12, 35, 0),
'cancelled': False,
'cancelled_at': None,
'cancelled_at_corrected': None,
'sub_total': 2000,
'delivery_fee': 250,
'total': 2250,
'_restaurant_id': 1,
'restaurant_notified_at': datetime.datetime(2020, 1, 2, 12, 5, 5),
'restaurant_notified_at_corrected': False,
'restaurant_confirmed_at': datetime.datetime(2020, 1, 2, 12, 5, 25),
'restaurant_confirmed_at_corrected': False,
'estimated_prep_duration': 900,
'estimated_prep_duration_corrected': False,
'estimated_prep_buffer': 480,
'_courier_id': 1,
'dispatch_at': datetime.datetime(2020, 1, 2, 12, 5, 1),
'dispatch_at_corrected': False,
'courier_notified_at': datetime.datetime(2020, 1, 2, 12, 6, 2),
'courier_notified_at_corrected': False,
'courier_accepted_at': datetime.datetime(2020, 1, 2, 12, 6, 17),
'courier_accepted_at_corrected': False,
'utilization': 50,
'_pickup_address_id': 1,
'reached_pickup_at': datetime.datetime(2020, 1, 2, 12, 16, 21),
'pickup_at': datetime.datetime(2020, 1, 2, 12, 18, 1),
'pickup_at_corrected': False,
'pickup_not_confirmed': False,
'left_pickup_at': datetime.datetime(2020, 1, 2, 12, 19, 45),
'left_pickup_at_corrected': False,
'_delivery_address_id': 2,
'reached_delivery_at': datetime.datetime(2020, 1, 2, 12, 27, 33),
'delivery_at': datetime.datetime(2020, 1, 2, 12, 29, 55),
'delivery_at_corrected': False,
'delivery_not_confirmed': False,
'_courier_waited_at_delivery': False,
'logged_delivery_distance': 500,
'logged_avg_speed': 7.89,
'logged_avg_speed_distance': 490,
}
@pytest.fixture
def order( # noqa:WPS211 pylint:disable=too-many-arguments
order_data, customer, restaurant, courier, address, address2,
):
"""An Order object."""
order = db.Order(**order_data)
order.customer = customer
order.restaurant = restaurant
order.courier = courier
order.pickup_address = address
order.delivery_address = address2
return order
@pytest.fixture
def restaurant_data():
"""The data for the Restaurant object."""
return {
'id': 1,
'created_at': datetime.datetime(2020, 1, 2, 3, 4, 5),
'name': 'Vevay',
'_address_id': 1,
'estimated_prep_duration': 1000,
}
@pytest.fixture
def restaurant(restaurant_data, address):
"""A Restaurant object."""
restaurant = db.Restaurant(**restaurant_data)
restaurant.address = address
return restaurant

141
tests/db/test_addresses.py Normal file
View file

@ -0,0 +1,141 @@
"""Test the ORM's Address model."""
import pytest
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in Address."""
# pylint:disable=no-self-use
def test_create_address(self, address_data):
"""Test instantiation of a new Address object."""
result = db.Address(**address_data)
assert result is not None
def test_text_representation(self, address_data):
"""Address has a non-literal text representation."""
address = db.Address(**address_data)
street = address_data['street']
city_name = address_data['city_name']
result = repr(address)
assert result == f'<Address({street} in {city_name})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in Address."""
# pylint:disable=no-self-use
def test_insert_into_database(self, address, db_session):
"""Insert an instance into the database."""
db_session.add(address)
db_session.commit()
def test_dublicate_primary_key(self, address, address_data, city, db_session):
"""Can only add a record once."""
db_session.add(address)
db_session.commit()
another_address = db.Address(**address_data)
another_address.city = city
db_session.add(another_address)
with pytest.raises(orm_exc.FlushError):
db_session.commit()
def test_delete_a_referenced_address(self, address, address_data, db_session):
"""Remove a record that is referenced with a FK."""
db_session.add(address)
db_session.commit()
# Fake a second address that belongs to the same primary address.
address_data['id'] += 1
another_address = db.Address(**address_data)
db_session.add(another_address)
db_session.commit()
with pytest.raises(sa_exc.IntegrityError):
db_session.execute(
db.Address.__table__.delete().where( # noqa:WPS609
db.Address.id == address.id,
),
)
def test_delete_a_referenced_city(self, address, city, db_session):
"""Remove a record that is referenced with a FK."""
db_session.add(address)
db_session.commit()
with pytest.raises(sa_exc.IntegrityError):
db_session.execute(
db.City.__table__.delete().where(db.City.id == city.id), # noqa:WPS609
)
@pytest.mark.parametrize('latitude', [-91, 91])
def test_invalid_latitude(self, address, db_session, latitude):
"""Insert an instance with invalid data."""
address.latitude = latitude
db_session.add(address)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
@pytest.mark.parametrize('longitude', [-181, 181])
def test_invalid_longitude(self, address, db_session, longitude):
"""Insert an instance with invalid data."""
address.longitude = longitude
db_session.add(address)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
@pytest.mark.parametrize('zip_code', [-1, 0, 9999, 100000])
def test_invalid_zip_code(self, address, db_session, zip_code):
"""Insert an instance with invalid data."""
address.zip_code = zip_code
db_session.add(address)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
@pytest.mark.parametrize('floor', [-1, 41])
def test_invalid_floor(self, address, db_session, floor):
"""Insert an instance with invalid data."""
address.floor = floor
db_session.add(address)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
class TestProperties:
"""Test properties in Address."""
# pylint:disable=no-self-use
def test_is_primary(self, address_data):
"""Test Address.is_primary property."""
address = db.Address(**address_data)
result = address.is_primary
assert result is True
def test_is_not_primary(self, address_data):
"""Test Address.is_primary property."""
address_data['_primary_id'] = 999
address = db.Address(**address_data)
result = address.is_primary
assert result is False

99
tests/db/test_cities.py Normal file
View file

@ -0,0 +1,99 @@
"""Test the ORM's City model."""
import pytest
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in City."""
# pylint:disable=no-self-use
def test_create_city(self, city_data):
"""Test instantiation of a new City object."""
result = db.City(**city_data)
assert result is not None
def test_text_representation(self, city_data):
"""City has a non-literal text representation."""
city = db.City(**city_data)
name = city_data['name']
result = repr(city)
assert result == f'<City({name})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in City."""
# pylint:disable=no-self-use
def test_insert_into_database(self, city, db_session):
"""Insert an instance into the database."""
db_session.add(city)
db_session.commit()
def test_dublicate_primary_key(self, city, city_data, db_session):
"""Can only add a record once."""
db_session.add(city)
db_session.commit()
another_city = db.City(**city_data)
db_session.add(another_city)
with pytest.raises(orm_exc.FlushError):
db_session.commit()
class TestProperties:
"""Test properties in City."""
# pylint:disable=no-self-use
def test_location_data(self, city_data):
"""Test City.location property."""
city = db.City(**city_data)
result = city.location
assert isinstance(result, dict)
assert len(result) == 2
assert result['latitude'] == pytest.approx(city_data['_center_latitude'])
assert result['longitude'] == pytest.approx(city_data['_center_longitude'])
def test_viewport_data_overall(self, city_data):
"""Test City.viewport property."""
city = db.City(**city_data)
result = city.viewport
assert isinstance(result, dict)
assert len(result) == 2
def test_viewport_data_northeast(self, city_data):
"""Test City.viewport property."""
city = db.City(**city_data)
result = city.viewport['northeast']
assert isinstance(result, dict)
assert len(result) == 2
assert result['latitude'] == pytest.approx(city_data['_northeast_latitude'])
assert result['longitude'] == pytest.approx(city_data['_northeast_longitude'])
def test_viewport_data_southwest(self, city_data):
"""Test City.viewport property."""
city = db.City(**city_data)
result = city.viewport['southwest']
assert isinstance(result, dict)
assert len(result) == 2
assert result['latitude'] == pytest.approx(city_data['_southwest_latitude'])
assert result['longitude'] == pytest.approx(city_data['_southwest_longitude'])

125
tests/db/test_couriers.py Normal file
View file

@ -0,0 +1,125 @@
"""Test the ORM's Courier model."""
import pytest
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in Courier."""
# pylint:disable=no-self-use
def test_create_courier(self, courier_data):
"""Test instantiation of a new Courier object."""
result = db.Courier(**courier_data)
assert result is not None
def test_text_representation(self, courier_data):
"""Courier has a non-literal text representation."""
courier_data['id'] = 1
courier = db.Courier(**courier_data)
id_ = courier_data['id']
result = repr(courier)
assert result == f'<Courier(#{id_})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in Courier."""
# pylint:disable=no-self-use
def test_insert_into_database(self, courier, db_session):
"""Insert an instance into the database."""
db_session.add(courier)
db_session.commit()
def test_dublicate_primary_key(self, courier, courier_data, db_session):
"""Can only add a record once."""
db_session.add(courier)
db_session.commit()
another_courier = db.Courier(**courier_data)
db_session.add(another_courier)
with pytest.raises(orm_exc.FlushError):
db_session.commit()
def test_invalid_vehicle(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.vehicle = 'invalid'
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_negative_speed(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.historic_speed = -1
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_unrealistic_speed(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.historic_speed = 999
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_negative_capacity(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.capacity = -1
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_too_much_capacity(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.capacity = 999
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_negative_pay_per_hour(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.pay_per_hour = -1
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_too_much_pay_per_hour(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.pay_per_hour = 9999
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_negative_pay_per_order(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.pay_per_order = -1
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_too_much_pay_per_order(self, courier, db_session):
"""Insert an instance with invalid data."""
courier.pay_per_order = 999
db_session.add(courier)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()

51
tests/db/test_customer.py Normal file
View file

@ -0,0 +1,51 @@
"""Test the ORM's Customer model."""
import pytest
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in Customer."""
# pylint:disable=no-self-use
def test_create_customer(self, customer_data):
"""Test instantiation of a new Customer object."""
result = db.Customer(**customer_data)
assert result is not None
def test_text_representation(self, customer_data):
"""Customer has a non-literal text representation."""
customer = db.Customer(**customer_data)
id_ = customer_data['id']
result = repr(customer)
assert result == f'<Customer(#{id_})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in Customer."""
# pylint:disable=no-self-use
def test_insert_into_database(self, customer, db_session):
"""Insert an instance into the database."""
db_session.add(customer)
db_session.commit()
def test_dublicate_primary_key(self, customer, customer_data, db_session):
"""Can only add a record once."""
db_session.add(customer)
db_session.commit()
another_customer = db.Customer(**customer_data)
db_session.add(another_customer)
with pytest.raises(orm_exc.FlushError):
db_session.commit()

397
tests/db/test_orders.py Normal file
View file

@ -0,0 +1,397 @@
"""Test the ORM's Order model."""
import datetime
import pytest
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in Order."""
# pylint:disable=no-self-use
def test_create_order(self, order_data):
"""Test instantiation of a new Order object."""
result = db.Order(**order_data)
assert result is not None
def test_text_representation(self, order_data):
"""Order has a non-literal text representation."""
order = db.Order(**order_data)
id_ = order_data['id']
result = repr(order)
assert result == f'<Order(#{id_})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in Order."""
# pylint:disable=no-self-use
def test_insert_into_database(self, order, db_session):
"""Insert an instance into the database."""
db_session.add(order)
db_session.commit()
def test_dublicate_primary_key(self, order, order_data, city, db_session):
"""Can only add a record once."""
db_session.add(order)
db_session.commit()
another_order = db.Order(**order_data)
another_order.city = city
db_session.add(another_order)
with pytest.raises(orm_exc.FlushError):
db_session.commit()
# TODO (order-constraints): the various Foreign Key and Check Constraints
# should be tested eventually. This is not of highest importance as
# we have a lot of confidence from the data cleaning notebook.
class TestProperties:
"""Test properties in Order."""
# pylint:disable=no-self-use,too-many-public-methods
def test_is_not_scheduled(self, order_data):
"""Test Order.scheduled property."""
order = db.Order(**order_data)
result = order.scheduled
assert result is False
def test_is_scheduled(self, order_data):
"""Test Order.scheduled property."""
order_data['ad_hoc'] = False
order_data['scheduled_delivery_at'] = datetime.datetime(2020, 1, 2, 12, 30, 0)
order_data['scheduled_delivery_at_corrected'] = False
order = db.Order(**order_data)
result = order.scheduled
assert result is True
def test_is_completed(self, order_data):
"""Test Order.completed property."""
order = db.Order(**order_data)
result = order.completed
assert result is True
def test_is_not_completed(self, order_data):
"""Test Order.completed property."""
order_data['cancelled'] = True
order_data['cancelled_at'] = datetime.datetime(2020, 1, 2, 12, 15, 0)
order_data['cancelled_at_corrected'] = False
order = db.Order(**order_data)
result = order.completed
assert result is False
def test_is_corrected(self, order_data):
"""Test Order.corrected property."""
order_data['dispatch_at_corrected'] = True
order = db.Order(**order_data)
result = order.corrected
assert result is True
def test_time_to_accept_no_dispatch_at(self, order_data):
"""Test Order.time_to_accept property."""
order_data['dispatch_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_accept)
def test_time_to_accept_no_courier_accepted(self, order_data):
"""Test Order.time_to_accept property."""
order_data['courier_accepted_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_accept)
def test_time_to_accept_success(self, order_data):
"""Test Order.time_to_accept property."""
order = db.Order(**order_data)
result = order.time_to_accept
assert isinstance(result, datetime.timedelta)
def test_time_to_react_no_courier_notified(self, order_data):
"""Test Order.time_to_react property."""
order_data['courier_notified_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_react)
def test_time_to_react_no_courier_accepted(self, order_data):
"""Test Order.time_to_react property."""
order_data['courier_accepted_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_react)
def test_time_to_react_success(self, order_data):
"""Test Order.time_to_react property."""
order = db.Order(**order_data)
result = order.time_to_react
assert isinstance(result, datetime.timedelta)
def test_time_to_pickup_no_reached_pickup_at(self, order_data):
"""Test Order.time_to_pickup property."""
order_data['reached_pickup_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_pickup)
def test_time_to_pickup_no_courier_accepted(self, order_data):
"""Test Order.time_to_pickup property."""
order_data['courier_accepted_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_pickup)
def test_time_to_pickup_success(self, order_data):
"""Test Order.time_to_pickup property."""
order = db.Order(**order_data)
result = order.time_to_pickup
assert isinstance(result, datetime.timedelta)
def test_time_at_pickup_no_reached_pickup_at(self, order_data):
"""Test Order.time_at_pickup property."""
order_data['reached_pickup_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_at_pickup)
def test_time_at_pickup_no_pickup_at(self, order_data):
"""Test Order.time_at_pickup property."""
order_data['pickup_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_at_pickup)
def test_time_at_pickup_success(self, order_data):
"""Test Order.time_at_pickup property."""
order = db.Order(**order_data)
result = order.time_at_pickup
assert isinstance(result, datetime.timedelta)
def test_scheduled_pickup_at_no_restaurant_notified( # noqa:WPS118
self, order_data,
):
"""Test Order.scheduled_pickup_at property."""
order_data['restaurant_notified_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.scheduled_pickup_at)
def test_scheduled_pickup_at_no_est_prep_duration(self, order_data): # noqa:WPS118
"""Test Order.scheduled_pickup_at property."""
order_data['estimated_prep_duration'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.scheduled_pickup_at)
def test_scheduled_pickup_at_success(self, order_data):
"""Test Order.scheduled_pickup_at property."""
order = db.Order(**order_data)
result = order.scheduled_pickup_at
assert isinstance(result, datetime.datetime)
def test_if_courier_early_at_pickup(self, order_data):
"""Test Order.courier_early property."""
order = db.Order(**order_data)
result = order.courier_early
assert bool(result) is True
def test_if_courier_late_at_pickup(self, order_data):
"""Test Order.courier_late property."""
# Opposite of test case before.
order = db.Order(**order_data)
result = order.courier_late
assert bool(result) is False
def test_if_restaurant_early_at_pickup(self, order_data):
"""Test Order.restaurant_early property."""
order = db.Order(**order_data)
result = order.restaurant_early
assert bool(result) is True
def test_if_restaurant_late_at_pickup(self, order_data):
"""Test Order.restaurant_late property."""
# Opposite of test case before.
order = db.Order(**order_data)
result = order.restaurant_late
assert bool(result) is False
def test_time_to_delivery_no_reached_delivery_at(self, order_data): # noqa:WPS118
"""Test Order.time_to_delivery property."""
order_data['reached_delivery_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_delivery)
def test_time_to_delivery_no_pickup_at(self, order_data):
"""Test Order.time_to_delivery property."""
order_data['pickup_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_to_delivery)
def test_time_to_delivery_success(self, order_data):
"""Test Order.time_to_delivery property."""
order = db.Order(**order_data)
result = order.time_to_delivery
assert isinstance(result, datetime.timedelta)
def test_time_at_delivery_no_reached_delivery_at(self, order_data): # noqa:WPS118
"""Test Order.time_at_delivery property."""
order_data['reached_delivery_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_at_delivery)
def test_time_at_delivery_no_delivery_at(self, order_data):
"""Test Order.time_at_delivery property."""
order_data['delivery_at'] = None
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='not set'):
int(order.time_at_delivery)
def test_time_at_delivery_success(self, order_data):
"""Test Order.time_at_delivery property."""
order = db.Order(**order_data)
result = order.time_at_delivery
assert isinstance(result, datetime.timedelta)
def test_courier_waited_at_delviery(self, order_data):
"""Test Order.courier_waited_at_delivery property."""
order_data['_courier_waited_at_delivery'] = True
order = db.Order(**order_data)
result = int(order.courier_waited_at_delivery.total_seconds())
assert result > 0
def test_courier_did_not_wait_at_delivery(self, order_data):
"""Test Order.courier_waited_at_delivery property."""
order_data['_courier_waited_at_delivery'] = False
order = db.Order(**order_data)
result = int(order.courier_waited_at_delivery.total_seconds())
assert result == 0
def test_if_delivery_early_success(self, order_data):
"""Test Order.delivery_early property."""
order_data['ad_hoc'] = False
order_data['scheduled_delivery_at'] = datetime.datetime(2020, 1, 2, 12, 30, 0)
order_data['scheduled_delivery_at_corrected'] = False
order = db.Order(**order_data)
result = order.delivery_early
assert bool(result) is True
def test_if_delivery_early_failure(self, order_data):
"""Test Order.delivery_early property."""
order = db.Order(**order_data)
with pytest.raises(AttributeError, match='scheduled'):
int(order.delivery_early)
def test_if_delivery_late_success(self, order_data):
"""Test Order.delivery_late property."""
order_data['ad_hoc'] = False
order_data['scheduled_delivery_at'] = datetime.datetime(2020, 1, 2, 12, 30, 0)
order_data['scheduled_delivery_at_corrected'] = False
order = db.Order(**order_data)
result = order.delivery_late
assert bool(result) is False
def test_if_delivery_late_failure(self, order_data):
"""Test Order.delivery_late property."""
order = db.Order(**order_data)
with pytest.raises(AttributeError, match='scheduled'):
int(order.delivery_late)
def test_no_total_time_for_pre_order(self, order_data):
"""Test Order.total_time property."""
order_data['ad_hoc'] = False
order_data['scheduled_delivery_at'] = datetime.datetime(2020, 1, 2, 12, 30, 0)
order_data['scheduled_delivery_at_corrected'] = False
order = db.Order(**order_data)
with pytest.raises(AttributeError, match='Scheduled'):
int(order.total_time)
def test_no_total_time_for_cancelled_order(self, order_data):
"""Test Order.total_time property."""
order_data['cancelled'] = True
order_data['cancelled_at'] = datetime.datetime(2020, 1, 2, 12, 15, 0)
order_data['cancelled_at_corrected'] = False
order = db.Order(**order_data)
with pytest.raises(RuntimeError, match='Cancelled'):
int(order.total_time)
def test_total_time_success(self, order_data):
"""Test Order.total_time property."""
order = db.Order(**order_data)
result = order.total_time
assert isinstance(result, datetime.timedelta)

80
tests/db/test_restaurants.py Normal file
View file

@ -0,0 +1,80 @@
"""Test the ORM's Restaurant model."""
import pytest
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import exc as orm_exc
from urban_meal_delivery import db
class TestSpecialMethods:
"""Test special methods in Restaurant."""
# pylint:disable=no-self-use
def test_create_restaurant(self, restaurant_data):
"""Test instantiation of a new Restaurant object."""
result = db.Restaurant(**restaurant_data)
assert result is not None
def test_text_representation(self, restaurant_data):
"""Restaurant has a non-literal text representation."""
restaurant = db.Restaurant(**restaurant_data)
name = restaurant_data['name']
result = repr(restaurant)
assert result == f'<Restaurant({name})>'
@pytest.mark.e2e
@pytest.mark.no_cover
class TestConstraints:
"""Test the database constraints defined in Restaurant."""
# pylint:disable=no-self-use
def test_insert_into_database(self, restaurant, db_session):
"""Insert an instance into the database."""
db_session.add(restaurant)
db_session.commit()
def test_dublicate_primary_key(self, restaurant, restaurant_data, db_session):
"""Can only add a record once."""
db_session.add(restaurant)
db_session.commit()
another_restaurant = db.Restaurant(**restaurant_data)
db_session.add(another_restaurant)
with pytest.raises(orm_exc.FlushError):
db_session.commit()
def test_delete_a_referenced_address(self, restaurant, address, db_session):
"""Remove a record that is referenced with a FK."""
db_session.add(restaurant)
db_session.commit()
with pytest.raises(sa_exc.IntegrityError):
db_session.execute(
db.Address.__table__.delete().where( # noqa:WPS609
db.Address.id == address.id,
),
)
def test_negative_prep_duration(self, restaurant, db_session):
"""Insert an instance with invalid data."""
restaurant.estimated_prep_duration = -1
db_session.add(restaurant)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()
def test_too_high_prep_duration(self, restaurant, db_session):
"""Insert an instance with invalid data."""
restaurant.estimated_prep_duration = 2500
db_session.add(restaurant)
with pytest.raises(sa_exc.IntegrityError):
db_session.commit()

8
tests/test_config.py
View file

@ -43,3 +43,11 @@ def test_no_database_uri_set(env, monkeypatch):
with pytest.warns(UserWarning, match='no DATABASE_URI'):
config_mod.get_config(env)
def test_random_testing_schema():
"""CLEAN_SCHEMA is randomized if not seti explicitly."""
result = config_mod.random_schema_name()
assert isinstance(result, str)
assert len(result) <= 10