"""Operational inventory contracts; frozen domain tests remain unchanged.""" import uuid from django.contrib.auth import get_user_model from django.db import connection from django.test import Client, TestCase from django.test.utils import CaptureQueriesContext from django.urls import reverse from apps.inventory import test_documents as fixtures, tests as inventory_fixture from apps.inventory import test_usage as usage_fixture, document_services as documents from apps.inventory import control_services as control, services as stock from apps.inventory.models import StockMovement, StockLedgerEntry, GoodsReceipt, StockCount from apps.parts import services as parts from .inventory_workspace import url, KINDS from .inventory_actions import action_url from .test_ui import capture_page class InventoryWorkspaceTests(fixtures.DocumentFixture): def setUp(self): self.client.force_login(self.actor) def confirm(self, kind, operation, row, data=None): target = action_url(kind, operation, row.pk if row else None) response = self.client.get(target) self.assertEqual(response.status_code, 200) capture_page(response, "inventory-" + kind + "-" + operation) return self.client.post(target, {"revision": response.context["revision"], **(data or {})}) def test_all_destinations_and_empty_states(self): for kind in ["overview", *KINDS]: response = self.client.get(url(kind)) self.assertEqual(response.status_code, 200, kind) self.assertContains(response, "Inventory overview") capture_page(response, "inventory-empty-" + kind) self.assertEqual(self.client.get(reverse("operations:parts_requests")).status_code, 200) def test_receiving_create_lines_post_and_replay(self): response = self.confirm("receipts", "create", None, {"destination": self.location.pk, "external_reference": "WORKSPACE-RECEIPT"}) self.assertEqual(response.status_code, 302) receipt = GoodsReceipt.objects.get(external_reference="WORKSPACE-RECEIPT") response = self.confirm("receipts", "lines", receipt, { "lines-TOTAL_FORMS": 1, "lines-INITIAL_FORMS": 0, "lines-0-spare_part": self.part.pk, "lines-0-quantity": 3}) self.assertEqual(response.status_code, 302) self.assertEqual(self.balance(), 0) target = action_url("receipts", "post", receipt.pk) confirmation = self.client.get(target) capture_page(confirmation, "inventory-receipt-confirmation") token = confirmation.context["revision"] self.assertEqual(self.client.post(target, {"revision": token}).status_code, 302) self.assertEqual(self.balance(), 3) self.assertEqual(self.client.post(target, {"revision": token}).status_code, 403) self.assertEqual(StockMovement.objects.count(), 1) response = self.client.get(url("receipts", receipt.pk)) self.assertContains(response, "POSTED") self.assertNotContains(response, action_url("receipts", "post", receipt.pk)) capture_page(response, "inventory-posted-receipt") def test_receipt_stale_confirmation_and_tampered_token(self): receipt = self.receipt() target = action_url("receipts", "post", receipt.pk) token = self.client.get(target).context["revision"] documents.set_goods_receipt_lines(actor=self.actor, receipt=receipt, expected_revision=fixtures.revision(receipt), lines=[dict(spare_part=self.part, quantity=5)]) for value in (token, "tampered"): response = self.client.post(target, {"revision": value}) self.assertEqual(response.status_code, 400) self.assertEqual(self.balance(), 0) def test_serial_receipt_duplicate_is_atomic_and_scoped(self): receipt = self.receipt(quantity=1, part=self.serial_part, identifiers=["VISIBLE-SERIAL"]) self.assertEqual(self.confirm("receipts", "post", receipt).status_code, 302) duplicate = self.receipt(quantity=1, part=self.serial_part, identifiers=["VISIBLE-SERIAL"]) self.assertEqual(self.confirm("receipts", "post", duplicate).status_code, 400) self.assertEqual(StockMovement.objects.count(), 1) self.assertContains(self.client.get(url("receipts", receipt.pk)), "VISIBLE-SERIAL") inventory_fixture.grant(self.staff, self.other_company, permission_names=["view_stock"]) self.client.force_login(self.staff) self.assertEqual(self.client.get(url("receipts", receipt.pk)).status_code, 404) self.assertNotContains(self.client.get(url("history")), "VISIBLE-SERIAL") def test_transfer_dispatch_and_receive_are_separate(self): self.receive(5) transfer = self.transfer(quantity=2) self.assertEqual(self.confirm("transfers", "dispatch", transfer).status_code, 302) self.assertEqual(self.balance(), 3) self.assertEqual(self.balance(self.destination), 0) response = self.client.get(url("transfers", transfer.pk)) self.assertContains(response, "DISPATCHED") capture_page(response, "inventory-dispatched-transfer") self.assertEqual(self.confirm("transfers", "receive", transfer).status_code, 302) self.assertEqual(self.balance(self.destination), 2) self.assertEqual(self.client.get(action_url("transfers", "dispatch", transfer.pk)).status_code, 403) def test_count_observation_does_not_post_until_reconciled(self): self.receive(5) count = control.create_stock_count(actor=self.actor, location=self.location, spare_part=self.part) self.assertEqual(self.confirm("counts", "start", count).status_code, 302) self.assertEqual(self.confirm("counts", "record", count, {"counted_quantity": 4, "note": "Observed four"}).status_code, 302) self.assertEqual(self.balance(), 5) response = self.client.get(url("counts", count.pk)) self.assertContains(response, "Variance") capture_page(response, "inventory-count-observation") self.assertEqual(self.confirm("counts", "reconcile", count).status_code, 302) self.assertEqual(self.balance(), 4) self.assertContains(self.client.get(url("adjustments")), "Approved physical count variance") capture_page(self.client.get(url("adjustments")), "inventory-adjustments") def test_operational_line_bound_rejects_oversize_without_calling_service(self): from unittest.mock import patch receipt = self.receipt() target = action_url("receipts", "lines", receipt.pk) token = self.client.get(target).context["revision"] data = {"revision": token, "lines-TOTAL_FORMS": 101, "lines-INITIAL_FORMS": 1} for index in range(101): data.update({f"lines-{index}-spare_part": self.part.pk, f"lines-{index}-quantity": 1}) with patch("apps.operations.inventory_actions.perform") as perform: response = self.client.post(target, data) self.assertEqual(response.status_code, 400) self.assertContains(response, "Please submit at most 100 forms", status_code=400) capture_page(response, "inventory-lines-error") perform.assert_not_called() self.assertEqual(receipt.lines.get(is_active=True).quantity, 3) self.assertFalse(StockMovement.objects.exists()) # The cap belongs to the formset, not a rewritten inventory service. from .inventory_actions import build_forms data["lines-TOTAL_FORMS"] = 100 _, lines = build_forms(self.actor, "receipts", "lines", receipt, data) self.assertTrue(lines.is_valid()) self.assertEqual(len(lines.forms), 100) def test_serialized_transfer_selection_and_posting_preserve_identity(self): from apps.inventory.models import SerializedStockUnit self.post(self.receipt(quantity=1, part=self.serial_part, identifiers=["REVIEW-SERIAL"])) unit = SerializedStockUnit.objects.get(identifier="REVIEW-SERIAL") transfer = documents.create_stock_transfer(actor=self.actor, source=self.location, destination=self.destination) response = self.confirm("transfers", "lines", transfer, {"lines-TOTAL_FORMS": 1, "lines-INITIAL_FORMS": 0, "lines-0-spare_part": self.serial_part.pk, "lines-0-quantity": 1, "lines-0-selected_units": [unit.pk]}) self.assertEqual(response.status_code, 302) self.assertEqual(self.confirm("transfers", "dispatch", transfer).status_code, 302) unit.refresh_from_db() self.assertEqual(unit.state, "IN_TRANSIT") self.assertEqual(self.confirm("transfers", "receive", transfer).status_code, 302) unit.refresh_from_db() self.assertEqual((unit.state, unit.current_location_id, unit.identifier), ("IN_STOCK", self.destination.pk, "REVIEW-SERIAL")) def test_view_permission_does_not_grant_mutation(self): receipt = self.receipt() inventory_fixture.grant(self.staff, self.company, permission_names=["view_stock"]) self.client.force_login(self.staff) response = self.client.get(url("receipts", receipt.pk)) self.assertEqual(response.status_code, 200) self.assertNotContains(response, action_url("receipts", "post", receipt.pk)) for method in (self.client.get, self.client.post): self.assertEqual(method(action_url("receipts", "post", receipt.pk)).status_code, 403) self.assertEqual(method(action_url("receipts", "create")).status_code, 403) def test_company_center_location_and_both_transfer_endpoints(self): own = self.receipt(destination=self.destination) foreign = self.receipt(destination=self.foreign) warehouse = self.receipt() transfer = self.transfer() inventory_fixture.grant(self.staff, self.company, center=self.center, permission_names=["view_stock", "receive_stock"]) self.client.force_login(self.staff) self.assertEqual(self.client.get(url("receipts", own.pk)).status_code, 200) for row in (foreign, warehouse): self.assertEqual(self.client.get(url("receipts", row.pk)).status_code, 404) self.assertEqual(self.client.post(action_url("receipts", "post", row.pk)).status_code, 404) self.assertEqual(self.client.get(url("transfers", transfer.pk)).status_code, 404) self.assertEqual(self.client.get(url("history"), {"location": self.foreign.pk}).status_code, 400) response = self.client.get(action_url("receipts", "create")) self.assertNotContains(response, str(self.foreign.pk)) self.assertNotContains(response, str(self.location.pk)) self.assertContains(response, str(self.destination.pk)) def test_department_scope_is_not_location_or_center_scope(self): from apps.access.services import create_role_assignment from apps.organization.assignment_services import create_assignment reader = get_user_model().objects.create_user(username="inventory-department") role = inventory_fixture.grant(self.staff, self.company, permission_names=["view_stock"]) assignment = create_assignment(user=reader, company=self.company, department=self.department) create_role_assignment(user=reader, role=role, organization_assignment=assignment) receipt = self.receipt(destination=self.destination) self.client.force_login(reader) self.assertEqual(self.client.get(url("receipts", receipt.pk)).status_code, 403) def test_sibling_center_and_permission_paths_are_isolated(self): from apps.organization.models import ServiceCenter sibling = ServiceCenter.objects.create(company=self.company, region=self.region, code="SIBLING", name="Sibling", center_type="OWN") location = stock.create_location(actor=self.actor, company=self.company, service_center=sibling, code="SIBLING", name="Sibling store", location_type="STORE") receipt = self.receipt(destination=location) inventory_fixture.grant(self.staff, self.company, center=self.center, permission_names=["view_stock", "receive_stock"]) inventory_fixture.grant(self.staff, self.company, center=sibling, permission_names=["receive_stock"]) self.client.force_login(self.staff) self.assertEqual(self.client.get(url("receipts", receipt.pk)).status_code, 404) self.assertEqual(self.client.post(action_url("receipts", "post", receipt.pk)).status_code, 404) self.assertNotContains(self.client.get(action_url("receipts", "create")), str(location.pk)) def test_signed_confirmation_cannot_be_borrowed_by_another_actor(self): receipt = self.receipt() target = action_url("receipts", "post", receipt.pk) token = self.client.get(target).context["revision"] inventory_fixture.grant(self.staff, self.company, permission_names=["view_stock", "receive_stock"]) self.client.force_login(self.staff) self.assertEqual(self.client.post(target, {"revision": token}).status_code, 400) self.assertEqual(self.balance(), 0) def test_create_transfer_and_edit_lines_use_existing_services(self): response = self.confirm("transfers", "create", None, {"source": self.location.pk, "destination": self.destination.pk}) self.assertEqual(response.status_code, 302) from apps.inventory.models import StockTransfer transfer = StockTransfer.objects.get() response = self.confirm("transfers", "lines", transfer, {"lines-TOTAL_FORMS": 1, "lines-INITIAL_FORMS": 0, "lines-0-spare_part": self.part.pk, "lines-0-quantity": 2}) self.assertEqual(response.status_code, 302) self.assertFalse(StockMovement.objects.exists()) self.assertEqual(self.confirm("transfers", "dispatch", transfer).status_code, 400) self.assertEqual(self.confirm("transfers", "cancel", transfer, {"reason": "Demand withdrawn"}).status_code, 302) def test_staff_alone_denied_and_login_required(self): self.client.force_login(self.staff) self.assertEqual(self.client.get(url()).status_code, 403) self.client.logout() self.assertEqual(self.client.get(url()).status_code, 302) def test_csrf_and_readonly_history(self): receipt = self.receipt() client = Client(enforce_csrf_checks=True) client.force_login(self.actor) self.assertEqual(client.post(action_url("receipts", "post", receipt.pk)).status_code, 403) self.assertEqual(self.client.post(url("history")).status_code, 405) self.assertEqual(StockMovement.objects.count(), 0) def test_history_filters_pagination_and_no_hidden_counterpart(self): for _ in range(27): self.receive(1) self.move(1) response = self.client.get(url("history"), {"part": self.part.pk, "location": self.location.pk}) self.assertEqual(len(response.context["records"]), 25) self.assertTrue(response.context["page"].has_next()) capture_page(response, "inventory-history") self.assertEqual(self.client.get(url("history"), {"part": "invalid"}).status_code, 400) self.assertEqual(self.client.get(url("history"), {"start": "2026-02-02", "end": "2026-01-01"}).status_code, 400) inventory_fixture.grant(self.staff, self.company, center=self.center, permission_names=["view_stock"]) self.client.force_login(self.staff) response = self.client.get(url("history")) self.assertContains(response, self.destination.code) self.assertNotContains(response, self.location.code) self.assertEqual(len(response.context["records"]), 1) def test_metrics_and_position_quantities(self): self.post(self.receipt(quantity=4)) self.transfer() control.create_stock_count(actor=self.actor, location=self.location, spare_part=self.part) response = self.client.get(url()) metrics = dict(response.context["metrics"]) self.assertEqual(metrics["Receipts posted today"], 1) self.assertEqual(metrics["Draft / dispatched transfers"], 1) self.assertEqual(metrics["Draft / active counts"], 1) capture_page(response, "inventory-overview") response = self.client.get(url("positions")) facts = dict(response.context["records"][0]["facts"]) self.assertEqual(facts["On hand"], 4) self.assertEqual(facts["Active reserved"], 0) capture_page(response, "inventory-positions") def test_direct_adjustment_requires_review_and_rejects_replay(self): self.receive(5) from .inventory_workspace import source position = source(self.actor, "positions").get(location=self.location, spare_part=self.part) target = reverse("operations:inventory_adjust", args=[position.pk]) response = self.client.get(target) self.assertEqual(response.status_code, 200) capture_page(response, "inventory-direct-adjustment") data = dict(revision=response.context["revision"], quantity_delta=-1, reason="LOSS", reference="FOUND-DAMAGE", note="Physical evidence") self.assertEqual(self.client.post(target, data).status_code, 302) self.assertEqual(self.client.post(target, data).status_code, 400) self.assertEqual(self.balance(), 4) def test_direct_adjustment_stale_stock_and_permissions(self): self.receive(5) from .inventory_workspace import source position = source(self.actor, "positions").get(location=self.location, spare_part=self.part) target = reverse("operations:inventory_adjust", args=[position.pk]) token = self.client.get(target).context["revision"] self.receive(1) data = dict(revision=token, quantity_delta=-1, reason="LOSS", reference="LOSS", note="Count evidence") self.assertEqual(self.client.post(target, data).status_code, 400) self.assertEqual(self.balance(), 6) inventory_fixture.grant(self.staff, self.company, permission_names=["view_stock"]) self.client.force_login(self.staff) self.assertEqual(self.client.post(target, data).status_code, 403) def test_search_and_scoped_location_filter(self): self.post(self.receipt()) self.assertEqual(len(self.client.get(url("positions"), {"search": "INV-PART", "location": self.location.pk}).context["records"]), 1) self.assertEqual(len(self.client.get(url("positions"), {"search": "absent"}).context["records"]), 0) def test_list_query_bounds_and_growth(self): inventory_fixture.grant(self.staff, self.company, permission_names=["view_stock"]) self.client.force_login(self.staff) def seed(index): part = parts.create_spare_part(category=self.category, part_code=f"GROW-{index}", name="Growth part", serialization_policy="NOT_SERIALIZED") self.post(self.receipt(quantity=3, part=part)) self.transfer(part=part) def measure(): result = {} for kind in ("positions", "history", "receipts", "transfers"): with CaptureQueriesContext(connection) as queries: response = self.client.get(url(kind)) self.assertEqual(response.status_code, 200) result[kind] = len(queries) self.assertLessEqual(len(queries), {"positions": 15, "history": 10, "receipts": 11, "transfers": 11}[kind], kind) capture_page(response, "inventory-populated-" + kind) return result seed(0) before = measure() for index in range(1, 9): seed(index) after = measure() self.assertEqual(before, after) print("Inventory workspace query counts:", after) class InventoryRequestWorkspaceTests(usage_fixture.UsageFixture, TestCase): @classmethod def setUpTestData(cls): usage_fixture.setup_usage(cls) def setUp(self): self.client.force_login(self.actor) def test_request_approval_and_job_links(self): row = self.request() response = self.client.get(url("requests", row.pk)) self.assertContains(response, self.case.job_number) capture_page(response, "inventory-request") target = action_url("requests", "approve", row.pk) token = self.client.get(target).context["revision"] self.assertEqual(self.client.post(target, {"revision": token}).status_code, 302) row.refresh_from_db() self.assertEqual(row.status, "APPROVED") self.assertFalse(StockMovement.objects.exists()) def test_inventory_permission_does_not_disclose_job(self): row = self.request() reader = get_user_model().objects.create_user(username="inventory-only-reader") inventory_fixture.grant(reader, self.company, center=self.center, permission_names=["view_stock", "approve_parts"]) self.client.force_login(reader) response = self.client.get(url("requests", row.pk)) self.assertEqual(response.status_code, 200) self.assertNotContains(response, self.case.job_number) self.assertNotContains(response, reverse("operations:detail", args=["cases", self.case.pk])) self.assertEqual(self.client.get(url("history"), {"case": self.case.pk}).status_code, 404) self.assertEqual(self.client.post(action_url("requests", "approve", row.pk)).status_code, 403) def test_custody_and_unused_return_are_distinct(self): issue = self.prepared_issue() from apps.service import test_repair as repair _, action = repair.prepared(self) self.consume(issue, action) self.return_unused(issue) response = self.client.get(url("issues")) self.assertContains(response, "Issued quantity") self.assertContains(response, "Custodian") capture_page(response, "inventory-custody") response = self.client.get(url("returns")) self.assertContains(response, "Returned unused") capture_page(response, "inventory-unused-return") self.assertEqual(len(response.context["records"]), 2) self.assertContains(response, "Consumed") def test_request_queries_do_not_grow_per_row(self): def read(): with CaptureQueriesContext(connection) as queries: response = self.client.get(url("requests")) self.assertEqual(response.status_code, 200) self.assertLessEqual(len(queries), 13) capture_page(response, "inventory-request-queue") return len(queries) self.request() before = read() for _ in range(8): self.request() self.assertEqual(before, read()) print("Inventory request queries:", before) def test_recovery_is_evidence_not_usable_stock_or_device_identifier_disclosure(self): from apps.service import test_repair as repair from apps.inventory import usage_services issue = self.prepared_issue() _, action = repair.prepared(self) replacement = self.consume(issue, action) quarantine = stock.create_location(actor=self.actor, company=self.company, code="QUARANTINE", name="Quarantine", location_type="QUARANTINE", service_center=self.center) usage_services.recover_defective_component(actor=self.engineer, repair_action=action, location=quarantine, component_description="Removed defective display", observed_identifier="RECOVERY-PRIVATE-ID", command_key=uuid.uuid4(), replacement=replacement) response = self.client.get(url("recoveries")) self.assertContains(response, "Removed defective display") self.assertNotContains(response, "RECOVERY-PRIVATE-ID") capture_page(response, "inventory-recovery") def test_request_crosslinks_and_readonly_case_filter(self): row = self.request() response = self.client.get(reverse("operations:job_section", args=[self.case.pk, "parts"])) self.assertContains(response, url("requests") + "?case=" + str(self.case.pk)) response = self.client.get(url("requests"), {"case": self.case.pk}) self.assertEqual(len(response.context["records"]), 1) self.assertContains(response, url("requests", row.pk)) def test_active_reservations_metric_does_not_claim_issue_eligibility(self): from apps.service import services as cases from apps.inventory import usage_services from django.core.exceptions import ValidationError self.receive() reservation = self.reserve(self.approve(self.request())) cases.cancel_service_case(service_case=self.case, cancelled_by=self.user, reason="Customer cancelled") reservation.refresh_from_db() self.assertEqual(reservation.status, "ACTIVE") response = self.client.get(url()) self.assertEqual(dict(response.context["metrics"])["Active reservations"], 1) self.assertNotContains(response, "Reservations awaiting issue") with self.assertRaises(ValidationError): usage_services.issue_reserved_parts(actor=self.actor, reservation=reservation, expected_revision=reservation.updated_at.isoformat()) self.release(reservation) self.assertEqual(dict(self.client.get(url()).context["metrics"])["Active reservations"], 0)