120 lines
5.6 KiB
Python
120 lines
5.6 KiB
Python
"""Независимые контрактные проверки; все БД создаются в TemporaryDirectory."""
|
|
import csv
|
|
from contextlib import closing
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
FIELDS = ['order_id', 'line_id', 'order_date', 'category', 'quantity', 'unit_price']
|
|
|
|
|
|
class PipelineTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.input = self.root / 'input.csv'
|
|
self.db = self.root / 'sales.sqlite'
|
|
self.reports = self.root / 'reports'
|
|
|
|
def write(self, rows):
|
|
with self.input.open('w', newline='', encoding='utf-8') as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(FIELDS)
|
|
writer.writerows(rows)
|
|
|
|
def run_etl(self, rows):
|
|
from src.pipeline import run_pipeline
|
|
self.write(rows)
|
|
run_pipeline(self.input, self.db, self.reports)
|
|
return json.loads((self.reports / 'quality.json').read_text(encoding='utf-8'))
|
|
|
|
def snapshot(self):
|
|
with closing(sqlite3.connect(self.db)) as con:
|
|
return tuple(con.execute(f'SELECT * FROM {table} ORDER BY 1,2').fetchall()
|
|
for table in ['facts', 'daily_category'])
|
|
|
|
def test_manual_fixture_cleaning_reasons_and_exact_money(self):
|
|
rows = [
|
|
[' A ', '1', '2026-01-01', ' HOME GOODS ', '2', '10.25'],
|
|
['A', '1', '01.01.2026', 'home goods', '2', '10.25'],
|
|
['A', '1', '2026-01-01', 'home goods', '3', '10.25'],
|
|
['B', '1', '2026/01/03', 'Books', '3', '0.10'],
|
|
['', '1', 'bad', '', '0', '-1'],
|
|
['C', '1', '01/02/2026', 'Books', '1', '1'],
|
|
['D', '1', '2026-02-30', 'Books', '1', '1'],
|
|
['E', '1', '2026-01-01', 'Books', '0', '1'],
|
|
['F', '1', '2026-01-01', 'Books', '1', '1.001'],
|
|
]
|
|
q = self.run_etl(rows)
|
|
self.assertEqual((q['input_rows'], q['accepted_rows'], q['rejected_rows']), (9, 2, 7))
|
|
self.assertEqual(q['revenue_kopecks'], 2080) # 2*1025 + 3*10
|
|
self.assertEqual(q['corrected_rows'], 2)
|
|
self.assertEqual({k: v for k,v in q['reasons'].items() if v}, {
|
|
'duplicate': 1, 'conflicting_key': 1, 'missing_required': 1,
|
|
'invalid_date': 2, 'invalid_quantity': 1, 'invalid_price': 1})
|
|
self.assertEqual(sum(q['reasons'].values()), q['rejected_rows'])
|
|
with closing(sqlite3.connect(self.db)) as con:
|
|
facts = con.execute('SELECT order_id,line_id,order_date,category,quantity,unit_price_kopecks,revenue_kopecks FROM facts ORDER BY 1,2').fetchall()
|
|
self.assertEqual(facts, [('A','1','2026-01-01','home goods',2,1025,2050), ('B','1','2026-01-03','books',3,10,30)])
|
|
self.assertEqual(con.execute('SELECT SUM(revenue_kopecks) FROM daily_category').fetchone()[0], 2080)
|
|
with (self.reports / 'rejected.csv').open(encoding='utf-8', newline='') as f:
|
|
rejected = list(csv.DictReader(f))
|
|
self.assertEqual([int(r['source_row']) for r in rejected], [3,4,6,7,8,9,10])
|
|
self.assertEqual([r['reason'] for r in rejected], ['duplicate','conflicting_key','missing_required','invalid_date','invalid_date','invalid_quantity','invalid_price'])
|
|
|
|
def test_first_valid_key_wins_and_lines_have_separate_keys(self):
|
|
q = self.run_etl([
|
|
['A','1','bad','x','1','1'],
|
|
['A','1','2026-01-01','x','1','1'],
|
|
['A','2','2026-01-01','x','1','2'],
|
|
])
|
|
self.assertEqual(q['accepted_rows'], 2)
|
|
self.assertEqual(q['revenue_kopecks'], 300)
|
|
|
|
def test_idempotency_logical_data(self):
|
|
rows = [['A','1','2026-01-01','x','2','0.10']]*2
|
|
first = self.run_etl(rows)
|
|
before = self.snapshot()
|
|
second = self.run_etl(rows)
|
|
self.assertEqual(before, self.snapshot())
|
|
self.assertEqual(first, second)
|
|
|
|
def test_empty_and_all_rejected_replace_old_snapshot(self):
|
|
valid = [['A','1','2026-01-01','x','1','1']]
|
|
for replacement in ([], [['','1','bad','x','0','0']]):
|
|
with self.subTest(replacement=replacement):
|
|
self.run_etl(valid)
|
|
q = self.run_etl(replacement)
|
|
self.assertEqual(self.snapshot(), ([], []))
|
|
self.assertEqual(q['accepted_rows'], 0)
|
|
self.assertEqual(q['revenue_kopecks'], 0)
|
|
|
|
def test_failed_write_rolls_back_previous_business_data(self):
|
|
from src.pipeline import run_pipeline
|
|
self.run_etl([['A','1','2026-01-01','x','1','1']])
|
|
before = self.snapshot()
|
|
with closing(sqlite3.connect(self.db)) as con:
|
|
con.execute("CREATE TRIGGER reject_insert BEFORE INSERT ON facts BEGIN SELECT RAISE(ABORT, 'injected write failure'); END")
|
|
self.write([['B','1','2026-01-02','y','2','2']])
|
|
with self.assertRaises(sqlite3.DatabaseError):
|
|
run_pipeline(self.input, self.db, self.reports)
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
def test_numeric_edge_cases_rejected_without_crash(self):
|
|
rows = []
|
|
for i, price in enumerate(['0','-1','NaN','Infinity','1e100','92233720368547758.08']):
|
|
rows.append([str(i),'1','2026-01-01','x','1',price])
|
|
for i, qty in enumerate(['-1','0','1.5','NaN']):
|
|
rows.append(['q'+str(i),'1','2026-01-01','x',qty,'1'])
|
|
q = self.run_etl(rows)
|
|
self.assertEqual(q['accepted_rows'], 0)
|
|
self.assertEqual(q['rejected_rows'], len(rows))
|
|
self.assertEqual(sum(q['reasons'].values()), len(rows))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|