from uuid import uuid4 from django.contrib.auth import get_user_model from django.db import connection from django.test import TestCase from django.test.utils import CaptureQueriesContext from apps.commercial.test_payment import PaymentFixture, setup_payment from apps.inventory.test_usage import setup_usage from apps.inventory import tests as stock_fixture, services as stock from apps.service import test_quality_control as qc, test_repair as repair from apps.reporting import commercial_analytics as commercial, inventory_analytics as inventory, service_analytics as service class WarrantyScenarioTests(PaymentFixture, TestCase): @classmethod def setUpTestData(cls): fixture = PaymentFixture() setup_payment(fixture, responsibility="WARRANTY") for key, value in vars(fixture).items(): setattr(cls, key, value) def test_warranty_invoice_has_no_customer_due_without_payment(self): row = commercial.settlements(self.actor, {}).get() self.assertEqual((row.customer_pay_total, row.warranty_covered_total, row.balance_due, row.clearance), (0, 100, 0, "NO_CUSTOMER_DUE")) self.assertFalse(commercial.payments(self.actor, {}).exists()) self.deliver() self.assertEqual(commercial.settlements(self.actor, {}).get().clearance, "NO_CUSTOMER_DUE") class MixedScenarioTests(PaymentFixture, TestCase): @classmethod def setUpTestData(cls): fixture = PaymentFixture() setup_usage(fixture) invoice = fixture.billed(mixed=True, consumed=2) customer = fixture.quote.lines.get(responsibility="CUSTOMER") warranty = fixture.quote.lines.get(responsibility="WARRANTY") invoice = fixture.reconcile(invoice, [fixture.allocation(line=customer), fixture.allocation(line=warranty)]) fixture.ready() fixture.invoice = fixture.finalize(invoice) for key, value in vars(fixture).items(): setattr(cls, key, value) def test_mixed_invoice_exact_split_and_partial_payment(self): row = commercial.settlements(self.actor, {}).get() self.assertEqual((row.grand_total, row.customer_pay_total, row.warranty_covered_total), (200, 100, 100)) self.pay(amount="40") row = commercial.settlements(self.actor, {}).get() self.assertEqual((row.balance_due, row.paid_amount), (60, 40)) summary = next(t for t in commercial.commercial_tables(self.actor, {}) if t.key == "invoice_values").rows.get() self.assertEqual((summary["value"], summary["customer"], summary["warranty"]), (200, 100, 100)) class InventoryScenarioTests(TestCase): @classmethod def setUpTestData(cls): stock_fixture.make_inventory_fixture(cls) stock.receive_stock(actor=cls.actor, destination=cls.location, spare_part=cls.part, quantity=5, reference="REPORT-NON-SERIAL", idempotency_key=uuid4()) unit = stock.register_serialized_unit(actor=cls.actor, company=cls.company, spare_part=cls.serial_part, identifier="REPORT-SERIAL") stock.receive_stock(actor=cls.actor, destination=cls.location, spare_part=cls.serial_part, quantity=1, units=[unit], reference="REPORT-SERIAL", idempotency_key=uuid4()) def test_multiple_parts_serialized_and_nonserialized(self): rows = {row.spare_part_id: row for row in inventory.positions(self.actor, {})} self.assertEqual((rows[self.part.pk].on_hand, rows[self.part.pk].nonserialized_quantity, rows[self.part.pk].serialized_units), (5, 5, 0)) self.assertEqual((rows[self.serial_part.pk].on_hand, rows[self.serial_part.pk].nonserialized_quantity, rows[self.serial_part.pk].serialized_units), (1, 0, 1)) def test_inventory_query_count_does_not_grow_with_positions(self): self.client.force_login(self.actor) with CaptureQueriesContext(connection) as initial_page: self.assertEqual(self.client.get("/reports/inventory/").status_code, 200) with CaptureQueriesContext(connection) as first: list(inventory.positions(self.actor, {}).values("location__code", "spare_part__name", "on_hand", "available", "serialized_units")) stock.move_stock(actor=self.actor, source=self.location, destination=self.destination, spare_part=self.part, quantity=2, reference="REPORT-MOVE", idempotency_key=uuid4()) with CaptureQueriesContext(connection) as second: list(inventory.positions(self.actor, {}).values("location__code", "spare_part__name", "on_hand", "available", "serialized_units")) self.assertEqual((len(first), len(second)), (1, 1)) with CaptureQueriesContext(connection) as final_page: self.assertEqual(self.client.get("/reports/inventory/").status_code, 200) self.assertEqual(len(initial_page), len(final_page)) def test_same_location_code_in_two_companies_does_not_merge_movements(self): self.assertEqual(self.location.code, self.foreign.code) stock.receive_stock(actor=self.actor, destination=self.foreign, spare_part=self.part, quantity=3, reference="REPORT-FOREIGN", idempotency_key=uuid4()) report = next(t for t in inventory.inventory_tables(self.actor, {}) if t.key == "movements") rows = list(report.rows.filter(spare_part_id=self.part.pk)) self.assertEqual(len(rows), 2) self.assertEqual({row["location_id"] for row in rows}, {self.location.pk, self.foreign.pk}) self.assertEqual({row["delta"] for row in rows}, {3, 5}) class ReworkScenarioTests(TestCase): @classmethod def setUpTestData(cls): fixture = type("Fixture", (), {})() qc.setup_qc(fixture) first = qc.prepared(fixture) qc.check(fixture, first, result="FAIL") qc.fail_qc(fixture, first) fixture.case.refresh_from_db() fixture.rework, _ = repair.prepared(fixture) repair.complete(fixture, fixture.rework) second = qc.prepared(fixture) qc.fill(fixture, second) qc.pass_qc(fixture, second) fixture.reader = get_user_model().objects.create_superuser(username="rework-report-reader") for key, value in vars(fixture).items(): setattr(cls, key, value) def test_qc_failure_and_rework_are_separate_authoritative_attempts(self): repair_rows = service.repairs(self.reader, {}) self.assertEqual(repair_rows.count(), 2) self.assertEqual(repair_rows.filter(after_failed_qc=True).count(), 1) self.assertEqual(repair_rows.filter(repeat_attempt=True).count(), 1) qc_rows = service.quality_controls(self.reader, {}) self.assertEqual(qc_rows.count(), 2) self.assertEqual(qc_rows.filter(first_completed_attempt=True).get().outcome, "FAILED") self.assertEqual(qc_rows.filter(outcome="PASSED").count(), 1) rates = service.service_rates(self.reader, {}) self.assertEqual([r["percent"] for r in rates], [100, 0, 50]) def test_failure_cooccurrence_groups_do_not_duplicate_qc(self): report = next(t for t in service.service_tables(self.reader, {}) if t.key == "qc_failure_actions") self.assertEqual(report.rows.get()["count"], 1)