311 lines
12 KiB
Python
311 lines
12 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
|
|
|
|
async def create_verified_center(client, owner_headers, admin_headers, internal_headers, name: str) -> dict:
|
|
center = (
|
|
await client.post(
|
|
"/api/service-centers",
|
|
headers=owner_headers,
|
|
json={"display_name": name, "country": "KR", "city": "Seoul"},
|
|
)
|
|
).json()
|
|
await client.post(
|
|
"/api/users",
|
|
headers=internal_headers,
|
|
json={"telegram_id": 9001, "platform_role": "admin"},
|
|
)
|
|
verified = await client.post(f"/api/admin/service-centers/{center['id']}/verify", headers=admin_headers)
|
|
assert verified.status_code == 200
|
|
return verified.json()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_employee_invite_activation_revoked_and_expired(
|
|
client, auth_headers, other_auth_headers, admin_auth_headers, internal_headers
|
|
) -> None:
|
|
center = await create_verified_center(client, auth_headers, admin_auth_headers, internal_headers, "Invite Flow Service")
|
|
|
|
invite = await client.post(
|
|
f"/api/service-centers/{center['id']}/employees/invite",
|
|
headers=auth_headers,
|
|
json={"telegram_id": 2002, "role": "manager"},
|
|
)
|
|
assert invite.status_code == 201 or invite.status_code == 200
|
|
employee = invite.json()
|
|
token = employee["invite_token"]
|
|
|
|
forbidden = await client.get(f"/api/sto/dashboard?service_center_id={center['id']}", headers=other_auth_headers)
|
|
assert forbidden.status_code == 403
|
|
|
|
accepted = await client.post(
|
|
f"/api/service-centers/employees/invites/{token}/accept",
|
|
headers=other_auth_headers,
|
|
)
|
|
assert accepted.status_code == 200
|
|
assert accepted.json()["status"] == "active"
|
|
|
|
dashboard = await client.get(f"/api/sto/dashboard?service_center_id={center['id']}", headers=other_auth_headers)
|
|
assert dashboard.status_code == 200
|
|
|
|
other_center = await create_verified_center(
|
|
client,
|
|
auth_headers,
|
|
admin_auth_headers,
|
|
internal_headers,
|
|
"Other Tenant Service",
|
|
)
|
|
cross_tenant_dashboard = await client.get(
|
|
f"/api/sto/dashboard?service_center_id={other_center['id']}",
|
|
headers=other_auth_headers,
|
|
)
|
|
assert cross_tenant_dashboard.status_code == 403
|
|
|
|
revoked_invite = (
|
|
await client.post(
|
|
f"/api/service-centers/{center['id']}/employees/invite",
|
|
headers=auth_headers,
|
|
json={"telegram_id": 3003, "role": "receptionist"},
|
|
)
|
|
).json()
|
|
revoked = await client.post(
|
|
f"/api/service-centers/employees/{revoked_invite['id']}/revoke-invite",
|
|
headers=auth_headers,
|
|
)
|
|
assert revoked.status_code == 200
|
|
assert revoked.json()["status"] == "revoked"
|
|
|
|
expired_invite = (
|
|
await client.post(
|
|
f"/api/service-centers/{center['id']}/employees/invite",
|
|
headers=auth_headers,
|
|
json={"telegram_id": 4004, "role": "mechanic", "expires_in_hours": 0},
|
|
)
|
|
).json()
|
|
expired_headers = {"X-Telegram-Init-Data": __import__("conftest").make_init_data(4004)}
|
|
expired = await client.post(
|
|
f"/api/service-centers/employees/invites/{expired_invite['invite_token']}/accept",
|
|
headers=expired_headers,
|
|
)
|
|
assert expired.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_work_order_completion_creates_vehicle_records_and_updates_costs(
|
|
client, auth_headers, other_auth_headers, admin_auth_headers, internal_headers
|
|
) -> None:
|
|
center = await create_verified_center(client, auth_headers, admin_auth_headers, internal_headers, "Work Order Service")
|
|
vehicle = (
|
|
await client.post(
|
|
"/api/my/vehicles",
|
|
headers=other_auth_headers,
|
|
json={"name": "WO car", "current_odometer": 10000},
|
|
)
|
|
).json()
|
|
await client.post(
|
|
f"/api/service-centers/{center['id']}/vehicle-links/owner-attach",
|
|
headers=other_auth_headers,
|
|
json={"car_id": vehicle["id"], "access_level": "full"},
|
|
)
|
|
start_at = datetime.now(UTC) + timedelta(days=3)
|
|
appointment = (
|
|
await client.post(
|
|
"/api/appointments",
|
|
headers=other_auth_headers,
|
|
json={
|
|
"service_center_id": center["id"],
|
|
"vehicle_id": vehicle["id"],
|
|
"service_type": "oil_change",
|
|
"service_name": "Oil change",
|
|
"requested_start_at": start_at.replace(hour=10, minute=0, second=0, microsecond=0).isoformat(),
|
|
"estimated_duration_minutes": 60,
|
|
"customer_comment": "Oil and filter",
|
|
},
|
|
)
|
|
).json()
|
|
confirmed = await client.post(
|
|
f"/api/sto/appointments/{appointment['id']}/confirm",
|
|
headers=auth_headers,
|
|
json={"comment": "Confirmed"},
|
|
)
|
|
assert confirmed.status_code == 200
|
|
|
|
work_order = (
|
|
await client.post(
|
|
f"/api/sto/appointments/{appointment['id']}/create-work-order",
|
|
headers=auth_headers,
|
|
json={"odometer": 10150},
|
|
)
|
|
).json()
|
|
assert work_order["status"] == "diagnosis"
|
|
|
|
labor = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/labor-items",
|
|
headers=auth_headers,
|
|
json={"work_type": "oil_change", "title": "Oil labor", "quantity": 1, "unit_price": 70},
|
|
)
|
|
product = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/product-items",
|
|
headers=auth_headers,
|
|
json={
|
|
"title": "Engine oil",
|
|
"category": "engine_oil",
|
|
"product_type": "engine_oil",
|
|
"quantity": 4,
|
|
"unit": "l",
|
|
"unit_price": 15,
|
|
"viscosity": "5W-30",
|
|
"used_volume": 4,
|
|
},
|
|
)
|
|
assert labor.status_code == 201
|
|
assert product.status_code == 201
|
|
|
|
submitted = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/submit-approval",
|
|
headers=auth_headers,
|
|
json={"comment": "Please approve"},
|
|
)
|
|
assert submitted.status_code == 200
|
|
assert submitted.json()["final_total"] == "130.00"
|
|
|
|
approved = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/approve",
|
|
headers=other_auth_headers,
|
|
json={"comment": "Approved"},
|
|
)
|
|
assert approved.status_code == 200
|
|
assert approved.json()["status"] == "approved_by_owner"
|
|
|
|
completed = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/complete",
|
|
headers=auth_headers,
|
|
json={"odometer": 10300},
|
|
)
|
|
assert completed.status_code == 200
|
|
assert completed.json()["status"] == "completed"
|
|
|
|
duplicate_completion = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/complete",
|
|
headers=auth_headers,
|
|
json={},
|
|
)
|
|
assert duplicate_completion.status_code == 200
|
|
assert duplicate_completion.json()["status"] == "completed"
|
|
|
|
correction = await client.post(
|
|
f"/api/work-orders/{work_order['id']}/corrections",
|
|
headers=auth_headers,
|
|
json={
|
|
"reason": "Typo in service comment",
|
|
"proposed_changes": {"service_comment": "Oil and filter replaced"},
|
|
"owner_approval_required": False,
|
|
},
|
|
)
|
|
assert correction.status_code == 201
|
|
assert correction.json()["created_version"] == completed.json()["version"]
|
|
corrections = await client.get(f"/api/work-orders/{work_order['id']}/corrections", headers=other_auth_headers)
|
|
assert corrections.status_code == 200
|
|
assert corrections.json()[0]["id"] == correction.json()["id"]
|
|
approved_correction = await client.post(
|
|
f"/api/work-orders/corrections/{correction.json()['id']}/approve",
|
|
headers=other_auth_headers,
|
|
json={"comment": "Correction accepted"},
|
|
)
|
|
assert approved_correction.status_code == 200
|
|
assert approved_correction.json()["status"] == "approved"
|
|
repeated_correction_decision = await client.post(
|
|
f"/api/work-orders/corrections/{correction.json()['id']}/reject",
|
|
headers=other_auth_headers,
|
|
json={"comment": "Too late"},
|
|
)
|
|
assert repeated_correction_decision.status_code == 409
|
|
|
|
service_history = await client.get(
|
|
f"/api/my/vehicles/{vehicle['id']}/service-history",
|
|
headers=other_auth_headers,
|
|
)
|
|
expenses = await client.get(f"/api/cars/{vehicle['id']}/expenses", headers=other_auth_headers)
|
|
refreshed = await client.get(f"/api/cars/{vehicle['id']}", headers=other_auth_headers)
|
|
stats = await client.get(
|
|
f"/api/cars/{vehicle['id']}/stats?date_from=2026-01-01&date_to=2099-12-31",
|
|
headers=other_auth_headers,
|
|
)
|
|
|
|
assert service_history.status_code == 200
|
|
assert any(item["id"] == work_order["id"] for item in service_history.json()["service_visits"])
|
|
assert sum(1 for item in service_history.json()["service_visits"] if item["id"] == work_order["id"]) == 1
|
|
assert len(expenses.json()) == 1
|
|
assert expenses.json()[0]["total_cost"] == "130.00"
|
|
assert refreshed.json()["current_odometer"] == 10300
|
|
assert refreshed.json()["engine_oil_type"] == "5W-30"
|
|
assert refreshed.json()["engine_oil_volume_l"] == "4.00"
|
|
assert stats.json()["total_cost"] == "130.00"
|
|
admin_notifications = await client.get("/api/admin/notifications?limit=100", headers=admin_auth_headers)
|
|
admin_events = {item["event_type"] for item in admin_notifications.json()["rows"]}
|
|
assert {"work_order_completed", "work_order_correction_requested", "work_order_correction_resolved"} <= admin_events
|
|
|
|
cannot_edit = await client.patch(
|
|
f"/api/work-orders/{work_order['id']}",
|
|
headers=auth_headers,
|
|
json={"diagnosis": "Changed"},
|
|
)
|
|
assert cannot_edit.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_blocks_ocr_after_threshold(client, auth_headers) -> None:
|
|
for _ in range(8):
|
|
response = await client.post(
|
|
"/api/ocr/vin",
|
|
headers=auth_headers,
|
|
files={"file": ("vin.txt", BytesIO(b"VIN KMHCT41BAHU123456"), "text/plain")},
|
|
)
|
|
assert response.status_code == 200
|
|
limited = await client.post(
|
|
"/api/ocr/vin",
|
|
headers=auth_headers,
|
|
files={"file": ("vin.txt", BytesIO(b"VIN KMHCT41BAHU123456"), "text/plain")},
|
|
)
|
|
assert limited.status_code == 429
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ocr_receipt_parser_extracts_date_and_fuel_fields(client, auth_headers) -> None:
|
|
response = await client.post(
|
|
"/api/ocr/parse-text-receipt",
|
|
headers=auth_headers,
|
|
files={
|
|
"file": (
|
|
"receipt.txt",
|
|
BytesIO(b"Shell 2026-05-01 total 120.00 40 l price 3.00"),
|
|
"text/plain",
|
|
)
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["entry_date"] == "2026-05-01"
|
|
assert payload["liters"] == "40"
|
|
assert payload["price_per_liter"] == "3.00"
|
|
assert payload["category"] == "fuel"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_security_headers_and_metrics(client, auth_headers) -> None:
|
|
blocked = await client.post(
|
|
"/api/ocr/vin",
|
|
headers=auth_headers,
|
|
files={"file": ("payload.exe", BytesIO(b"MZ fake binary"), "application/octet-stream")},
|
|
)
|
|
assert blocked.status_code == 415
|
|
assert blocked.headers["x-content-type-options"] == "nosniff"
|
|
assert blocked.headers["referrer-policy"] == "strict-origin-when-cross-origin"
|
|
assert "x-request-id" in blocked.headers
|
|
|
|
metrics = await client.get("/metrics")
|
|
assert metrics.status_code == 200
|
|
assert "carpass_requests_total" in metrics.text
|