56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
import sqlite3
|
|
from contextlib import closing
|
|
|
|
import pytest
|
|
|
|
from src.database import SCHEMA, reload_database, save_query_results
|
|
from src.transform import Order
|
|
|
|
|
|
def make_database(tmp_path, sales):
|
|
path = tmp_path / "orders.db"
|
|
orders = [
|
|
Order(i, f"2024-01-{day:02}", category, 1, amount, amount, i)
|
|
for i, (day, category, amount) in enumerate(sales, 1)
|
|
]
|
|
reload_database(path, orders)
|
|
return path
|
|
|
|
|
|
def query(path, name):
|
|
with closing(sqlite3.connect(path)) as db:
|
|
return db.execute(SCHEMA.with_name(name + ".sql").read_text(encoding="utf-8")).fetchall()
|
|
|
|
|
|
def test_ranking_and_calendar_gap(tmp_path):
|
|
db = make_database(tmp_path, [(1, "B", 50), (1, "B", 50), (3, "A", 100), (3, "C", 1)])
|
|
assert query(db, "top_categories") == [("A", 1, 100), ("B", 2, 100), ("C", 1, 1)]
|
|
assert query(db, "daily_revenue") == [
|
|
("2024-01-01", 100, None), ("2024-01-02", 0, -100), ("2024-01-03", 101, 101),
|
|
]
|
|
|
|
|
|
def test_separate_categories_and_zero_days(tmp_path):
|
|
sales = [(day, "Постоянная", 50) for day in range(1, 8)]
|
|
sales += [(day, "Падение", 100) for day in range(1, 8) if day != 4]
|
|
sales += [(4, "Рост", 100)]
|
|
db = make_database(tmp_path, sales)
|
|
assert query(db, "anomalies") == [
|
|
("2024-01-04", "Падение", 0), ("2024-01-04", "Рост", 100),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("amounts", [[10] * 7, [1] * 5 + [100], [10] * 8 + [20] * 2])
|
|
def test_constant_short_or_exactly_two_standard_deviations(tmp_path, amounts):
|
|
db = make_database(tmp_path, [(i, "A", value) for i, value in enumerate(amounts, 1)])
|
|
assert query(db, "anomalies") == []
|
|
|
|
|
|
def test_empty_queries_and_report(tmp_path):
|
|
db = make_database(tmp_path, [])
|
|
for name in ("top_categories", "daily_revenue", "anomalies"):
|
|
assert query(db, name) == []
|
|
destination = tmp_path / "query_results.md"
|
|
save_query_results(db, destination)
|
|
assert "Аномалий не обнаружено." in destination.read_text(encoding="utf-8")
|