diff --git a/src/testing/integration_test.py b/src/testing/integration_test.py new file mode 100644 index 0000000..8f4dca4 --- /dev/null +++ b/src/testing/integration_test.py @@ -0,0 +1,905 @@ +""" +Integration tests for Optimum OH backend. +All endpoints are sourced from frontend API calls using OPTIMUM_OH_API_URL. + +Run with: + pytest src/testing/integration_test.py -v +""" + +import pytest +import requests + +BASE_URL = "http://100.125.115.116:8000/optimumoh" +AUTH_BASE_URL = "http://100.125.115.116:8000/auth" + +# Replace with a valid token before running +AUTH_TOKEN = "Bearer " + +HEADERS = { + "Authorization": AUTH_TOKEN, + "Content-Type": "application/json", +} + +# Placeholder IDs — replace with real values from your database before running +OVERHAUL_SESSION_ID = 1 +OVERHAUL_SCHEDULE_ID = 1 +OVERHAUL_EQUIPMENT_ID = 1 +SCOPE_JOB_ID = 1 +CALCULATION_ID = 1 +BUDGET_THRESHOLD = 100000 +RISK_COST = 1 +LOCATION_TAG = "LOCATION-001" +SCOPE_NAME = "scope-name-example" +OVERHAUL_SESSION_HISTORY_ID = 1 +ASSET_NUM = "ASSET-001" +SCOPE_EQUIPMENT_JOB_ID = 1 +EQUIPMENT_ACTIVITY_ID = 1 +EAF_INPUT = 0.85 +DURATION = 8760 +SCOPE_EQUIPMENT_ACTIVITY_ID = 1 +ITEMNUM = "ITEM-001" +SPREADSHEET_LINK = "https://docs.google.com/spreadsheets/d/example_spreadsheet_id/edit" +# UUID-format placeholder — replace with a real UUID from your database +OVERHAUL_SESSION_UUID = "00000000-0000-0000-0000-000000000001" + +# --------------------------------------------------------------------------- +# Users +# --------------------------------------------------------------------------- +USER_LIST = [ + {"id": 1, "name": "devmustbeadmin", "pass": "", "email": "devmustbeadmin@mail.com", "authorization_token": ""}, + # {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "devmustbeengineer@mail.com", "authorization_token": ""}, + # {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "devmustbeapplicationadmin@mail.com", "authorization_token": ""}, + # {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "devmustbemanagement@mail.com", "authorization_token": ""}, +] + +# --------------------------------------------------------------------------- +# Role-based access constants +# --------------------------------------------------------------------------- +# Endpoints guarded by require_any_role(*ALLOWED_ROLES) block Management entirely. +# Endpoints with only required_permission(READ) allow Management (has READ in ACL). +# Endpoints with no guard at all allow every role. +NON_MANAGEMENT = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin"} +ALL_ALLOWED = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin", "devmustbemanagement"} + + +def user_auth_headers(token: str) -> dict: + return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + +def get_csrf_token(path: str, auth_token: str) -> str: + """Fetch a one-time CSRF token from be-auth for the given endpoint path. + + Args: + path: The endpoint path as seen by be-optimumoh (e.g. "/spareparts"). + Do NOT include the "/optimumoh" Kong prefix. + auth_token: The raw JWT token string (without "Bearer " prefix). + + Returns: + The plain-text CSRF token to send in the X-CSRF-Token header. + """ + res = requests.post( + f"{AUTH_BASE_URL}/csrf-token", + headers={ + "Authorization": f"Bearer {auth_token}", + "Content-Type": "application/json", + "X-Target-Path": path, + }, + json={ + "_csrf_request": True, + } + ) + print(f"Requested CSRF token for {path}, got status {res.status_code}, response: {res.text}") + assert res.status_code == 200, f"Failed to get CSRF token for {path}: {res.status_code} {res.text}" + data = res.json() + # Response shape: {"data": {"csrf_token": "..."}, ...} + return data["data"]["csrf_token"] + + +# ───────────────────────────────────────────── +# Overhaul Session +# ───────────────────────────────────────────── + +# # Note: Source code got commented +# def test_get_overhaul_session(): +# """GET /overhaul-session?all=1 — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get(f"{BASE_URL}/overhaul-session?all=1", headers=user_auth_headers(user["authorization_token"])) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# print(f"[{user['name']}] expected {expected}, got {res.status_code}, response: {res.text}") +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + + +# # Note: Source code got commented +# def test_post_overhaul_session(): +# """POST /overhaul-session — require_any_role + CREATE blocks Management""" +# payload = { +# "type": "Major", +# "start_date": "2025-01-01", +# "end_date": "2025-12-31", +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/overhaul-session", json=payload, headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + + + +# # ───────────────────────────────────────────── +# # Overhaul Schedules +# # ───────────────────────────────────────────── + + +# def test_post_overhaul_schedules(): +# """POST /overhaul-schedules — require_any_role + CREATE blocks Management""" +# payload = { +# "start_date": "2024-06-01T00:00:00Z", +# "end_date": "2024-06-30T23:59:59Z", +# "duration_oh": 720, +# "crew_number": 10, +# "status": "Upcoming" +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/overhaul-schedules", json=payload, headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_delete_overhaul_schedule(): +# """DELETE /overhaul-schedules/{id} — require_any_role + DELETE blocks Management""" +# for user in USER_LIST: +# res = requests.delete( +# f"{BASE_URL}/overhaul-schedules/{OVERHAUL_SCHEDULE_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + + + +# ───────────────────────────────────────────── +# Overhaul Activity +# ───────────────────────────────────────────── + +# def test_post_overhaul_activity_current(): +# """POST /overhaul-activity/{session_id} — require_any_role blocks Management""" +# payload = { +# "location_tag": LOCATION_TAG, +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_delete_overhaul_activity(): +# """DELETE /overhaul-activity/{overhaul_session}/delete/{location_tag} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.delete( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}/delete/{LOCATION_TAG}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# ───────────────────────────────────────────── +# Overhaul Jobs +# ───────────────────────────────────────────── + + +# def test_post_overhaul_jobs(): +# """POST /overhaul-jobs/{scope_job_id} — no role guard; all roles allowed""" +# payload = { +# "overhaul_equipment_id": OVERHAUL_EQUIPMENT_ID, +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-jobs/{SCOPE_JOB_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + + +# ───────────────────────────────────────────── +# Scope Equipments +# ───────────────────────────────────────────── + + +# def test_post_scope_equipments(): +# """POST /scope-equipments — require_any_role blocks Management""" +# payload = { +# "assetnums": ["A26312", "A26313", "A26314"], +# "scope_name": "A" +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/scope-equipments", json=payload, headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_delete_scope_equipment(): +# """DELETE /scope-equipments/delete/{id} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.delete( +# f"{BASE_URL}/scope-equipments/delete/{OVERHAUL_EQUIPMENT_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# ───────────────────────────────────────────── +# Scope Equipment Jobs +# ───────────────────────────────────────────── + +# def test_post_scope_equipment_jobs(): +# """POST /scope-equipment-jobs/{assetnum} — no role guard; all roles allowed""" +# payload = { +# "job_id": SCOPE_JOB_ID, +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/scope-equipment-jobs/{ASSET_NUM}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + + +# def test_delete_scope_equipment_job(): +# """DELETE /scope-equipment-jobs/delete/{id} — no role guard; all roles allowed""" +# for user in USER_LIST: +# res = requests.delete( +# f"{BASE_URL}/scope-equipment-jobs/delete/{SCOPE_EQUIPMENT_JOB_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Calculation — Budget Constraint +# # ───────────────────────────────────────────── + +# def test_post_calculation_time_constraint(): +# """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE""" +# payload = { +# # "overhaulCost": 500000000, +# # "overhaul_session_id": "86d6610e-1d44-4221-a521-e1dff2ea150f", +# # "costPerFailure": 0 +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_post_calculation_time_constraint_select_equipment(): +# """POST /calculation/time-constraint?scope_calculation_id=... — required_permission(CREATE); Management has no CREATE""" +# payload = {} +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint?scope_calculation_id={CALCULATION_ID}&with_results=1", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_put_calculation_time_constraint_update(): +# """PUT /calculation/time-constraint/update/{calculationId} — required_permission(EDIT); Management has no EDIT""" +# payload = { +# "risk_cost": RISK_COST, +# } +# for user in USER_LIST: +# res = requests.put( +# f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + +# ───────────────────────────────────────────── +# Spareparts +# ───────────────────────────────────────────── + + +# def test_post_spareparts(): +# """POST /spareparts — require_any_role blocks Management""" +# payload = { +# "name": "Test Part", +# "part_number": "TP-001", +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/spareparts", json=payload, headers=user_auth_headers(user["authorization_token"])) +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Equipment Activities / Workscopes +# # ───────────────────────────────────────────── + +# def test_get_equipment_workscopes(): +# """GET /equipment-workscopes/{location_tag} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/equipment-workscopes/{LOCATION_TAG}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + + +# def test_post_equipment_activities(): +# """POST /equipment-activities — no role guard; all roles allowed""" +# payload = { +# "location_tag": LOCATION_TAG, +# "workscope": "Inspection", +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/equipment-activities", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + + +# def test_delete_equipment_activity(): +# """DELETE /equipment-activities/delete/{id} — no role guard; all roles allowed""" +# for user in USER_LIST: +# res = requests.delete( +# f"{BASE_URL}/equipment-activities/delete/{EQUIPMENT_ACTIVITY_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Jobs +# # ───────────────────────────────────────────── + +# def test_get_jobs(): +# """GET /jobs — no role guard; all roles allowed""" +# for user in USER_LIST: +# res = requests.get(f"{BASE_URL}/jobs", headers=user_auth_headers(user["authorization_token"])) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# # ───────────────────────────────────────────── +# # Overhaul Jobs +# # ───────────────────────────────────────────── + +# def test_get_overhaul_jobs(): +# """GET /overhaul-jobs/{overhaul_equipment_id} — no role guard; all roles allowed""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/overhaul-jobs/{OVERHAUL_EQUIPMENT_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# # ───────────────────────────────────────────── +# # Calculation — Time Constraint +# # ───────────────────────────────────────────── + +# def test_get_calculation_time_constraint_by_id(): +# """GET /calculation/time-constraint/{calculationId} — required_permission(READ) only; Management has READ""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}?risk_cost={RISK_COST}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# def test_get_calculation_time_constraint_parameters(): +# """GET /calculation/time-constraint/parameters — required_permission(READ) only; Management has READ""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint/parameters", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# # ───────────────────────────────────────────── +# # Calculation — Target Reliability +# # ───────────────────────────────────────────── + +# def test_get_calculation_target_reliability(): +# """GET /calculation/target-reliability — required_permission(READ) only; Management has READ""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/target-reliability" +# f"?eaf_input={EAF_INPUT}&oh_session_id={OVERHAUL_SESSION_ID}&duration={DURATION}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# def test_get_calculation_target_reliability_with_cut_hours(): +# """GET /calculation/target-reliability?...&cut_hours=... — required_permission(READ) only; Management has READ""" +# cut_hours = 500 +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/target-reliability" +# f"?oh_session_id={OVERHAUL_SESSION_ID}&eaf_input={EAF_INPUT}" +# f"&duration={DURATION}&cut_hours={cut_hours}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# # ───────────────────────────────────────────── +# # Overhaul Activity +# # ───────────────────────────────────────────── + +# def test_get_overhaul_activity(): +# """GET /overhaul-activity/{overhaul_session} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + + +# # ============================================================================= +# # NEW TESTS — endpoints missing from the original test file +# # ============================================================================= + +# # ───────────────────────────────────────────── +# # Overhauls sub-routes +# # ───────────────────────────────────────────── + +# def test_get_overhauls_schedules(): +# """GET /overhauls/schedules — required_permission(READ); Management has READ""" +# for user in USER_LIST: +# res = requests.get(f"{BASE_URL}/overhauls/schedules", headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# def test_get_overhauls_critical_parts(): +# """GET /overhauls/critical-parts — required_permission(READ); Management has READ""" +# for user in USER_LIST: +# res = requests.get(f"{BASE_URL}/overhauls/critical-parts", headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# # ───────────────────────────────────────────── +# # Scope Equipments +# # ───────────────────────────────────────────── + +# # Could not test endpoint due to unknown oh_session_id +# def test_get_scope_equipments_history(): +# """GET /scope-equipments/history/{oh_session_id} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/scope-equipments/history/{OVERHAUL_SESSION_UUID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + + +# def test_post_scope_equipments(): +# """POST /scope-equipments — require_any_role blocks Management +# Payload: assetnums (List[str]), scope_name (str), removal_date (optional datetime), type (optional str) +# """ +# payload = { +# "assetnums": [ASSET_NUM], +# "scope_name": SCOPE_NAME, +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/scope-equipments", json=payload, headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Overhaul Activity +# # ───────────────────────────────────────────── + +# def test_get_overhaul_activity(): +# """GET /overhaul-activity/{overhaul_session} — require_any_role blocks Management +# Query params: location_tag (optional), scope_name (optional) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + + +# def test_post_overhaul_activity(): +# """POST /overhaul-activity/{overhaul_session_id} — require_any_role blocks Management +# Payload: location_tags (List[str]) +# Path: overhaul_session_id must be a valid UUID +# """ +# payload = { +# "location_tags": [LOCATION_TAG], +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_get_overhaul_activity_by_assetnum(): +# """GET /overhaul-activity/{overhaul_session}/{assetnum} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}/{ASSET_NUM}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + + +# def test_post_overhaul_activity_delete(): +# """POST /overhaul-activity/delete/{overhaul_session}/{location_tag} — require_any_role blocks Management +# No body required. CSRF token required in production. +# Path: overhaul_session must be a valid UUID +# """ +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-activity/delete/{OVERHAUL_SESSION_UUID}/{LOCATION_TAG}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Workscopes +# # ───────────────────────────────────────────── + +# def test_post_workscopes(): +# """POST /workscopes — no role guard; all roles allowed +# Payload: workscope (str) +# """ +# payload = { +# # "workscope": "Test workscope description", +# # "system": "Test system", +# # "subsystem": "Test subsystem", +# # "tasks": "Test task", +# # "equipment_workscope_groups": "Test group", +# # "oh_types": "Test OH type", +# "description": "Test description", +# # "job_id": SCOPE_JOB_ID, +# } +# for user in USER_LIST: +# res = requests.post(f"{BASE_URL}/workscopes", json=payload, headers=user_auth_headers(user["authorization_token"])) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + + +# def test_get_workscopes_by_id(): +# """GET /workscopes/{scope_equipment_activity_id} — no role guard; all roles allowed""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/workscopes/{SCOPE_EQUIPMENT_ACTIVITY_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}" + + +# def test_post_workscopes_update(): +# """POST /workscopes/update/{scope_equipment_activity_id} — no role guard; all roles allowed +# Payload: description (str) +# """ +# payload = { +# "description": "Updated workscope description", +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/workscopes/update/{SCOPE_EQUIPMENT_ACTIVITY_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}" + + +# def test_post_workscopes_delete(): +# """POST /workscopes/delete/{scope_equipment_activity_id} — no role guard; all roles allowed +# No body required. +# """ +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/workscopes/delete/{SCOPE_EQUIPMENT_ACTIVITY_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Equipment Workscopes +# # ───────────────────────────────────────────── + +# def test_get_equipment_workscopes(): +# """GET /equipment-workscopes/{location_tag} — no role guard; all roles allowed +# Query params: scope (optional str) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/equipment-workscopes/{LOCATION_TAG}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# def test_post_equipment_workscopes(): +# """POST /equipment-workscopes/{assetnum} — no role guard; all roles allowed +# Payload: job_ids (List[UUID]) +# """ +# payload = { +# "job_ids": [], +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/equipment-workscopes/{ASSET_NUM}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + + +# def test_post_equipment_workscopes_delete(): +# """POST /equipment-workscopes/delete/{scope_job_id} — no role guard; all roles allowed +# No body required. +# """ +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/equipment-workscopes/delete/{SCOPE_JOB_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Overhaul Schedules (additional) +# # ───────────────────────────────────────────── + +# def test_get_overhaul_schedules_by_id(): +# """GET /overhaul-schedules/{overhaul_session_id} — require_any_role blocks Management""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/overhaul-schedules/{OVERHAUL_SESSION_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + + +# def test_post_overhaul_schedules_update(): +# """POST /overhaul-schedules/update/{scope_id} — require_any_role + EDIT blocks Management +# Payload: duration_oh (optional int), crew_number (optional int), status (optional str) +# """ +# payload = { +# "duration_oh": 720, +# "crew_number": 10, +# "status": "Upcoming", +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-schedules/update/{OVERHAUL_SESSION_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_post_overhaul_schedules_delete(): +# """POST /overhaul-schedules/delete/{scope_id} — require_any_role + DELETE blocks Management +# No body required. CSRF token required in production. +# """ +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/overhaul-schedules/delete/{OVERHAUL_SESSION_ID}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Calculation — Time Constraint +# # ───────────────────────────────────────────── + +# def test_post_calculation_time_constraint(): +# """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE +# Payload: overhaulCost (float), ohSessionId (UUID), costPerFailure (float) +# Query params: scope_calculation_id (optional), with_results (optional int), simulation_id (optional) +# """ +# payload = { +# "overhaulCost": 0, +# "ohSessionId": OVERHAUL_SESSION_UUID, +# "costPerFailure": 0, +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# def test_get_calculation_time_constraint(): +# """GET /calculation/time-constraint — required_permission(READ); Management has READ""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# def test_get_calculation_time_constraint_parameters(): +# """GET /calculation/time-constraint/parameters — required_permission(READ); Management has READ +# Query params: calculation_id (optional) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint/parameters", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# def test_get_calculation_time_constraint_by_id(): +# """GET /calculation/time-constraint/{calculation_id} — required_permission(READ); Management has READ +# Query params: risk_cost (optional int, default 1) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}?risk_cost={RISK_COST}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}" + + +# def test_get_calculation_time_constraint_by_id_and_assetnum(): +# """GET /calculation/time-constraint/{calculation_id}/{assetnum} — required_permission(READ); Management has READ""" +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}/{ASSET_NUM}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}" + + +# def test_post_calculation_time_constraint_simulation(): +# """POST /calculation/time-constraint/{calculation_id}/simulation — required_permission(READ); Management has READ +# Payload: intervalDays (int) +# """ +# payload = { +# "intervalDays": 365, +# } +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}/simulation", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# assert res.status_code in (200, 201, 404), f"[{user['name']}] expected 200/201/404, got {res.status_code}" + + +# def test_post_calculation_time_constraint_update(): +# """POST /calculation/time-constraint/update/{calculation_id} — required_permission(EDIT); Management has no EDIT +# Payload: List of { id (UUID), is_included (bool) }. CSRF token required in production. +# """ +# payload = [] +# for user in USER_LIST: +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}", +# json=payload, +# headers=user_auth_headers(user["authorization_token"]), +# ) +# print(f"[{user['name']}], got {res.status_code}, response: {res.text}") +# if user["name"] in NON_MANAGEMENT: +# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" +# else: +# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + + +# # ───────────────────────────────────────────── +# # Calculation — Target Reliability +# # ───────────────────────────────────────────── + +# def test_get_calculation_target_reliability(): +# """GET /calculation/target-reliability — required_permission(READ); Management has READ +# Query params: oh_session_id (required), eaf_input (float), duration (int), simulation_id (optional), cut_hours (optional) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/target-reliability" +# f"?oh_session_id={OVERHAUL_SESSION_UUID}&eaf_input={EAF_INPUT}&duration={DURATION}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# assert res.status_code in (200, 400, 404, 425), f"[{user['name']}] expected 200/400/404/425, got {res.status_code}: {res.text}" + + +# # ───────────────────────────────────────────── +# # Calculation — Budget Constraint +# # ───────────────────────────────────────────── + +# def test_get_calculation_budget_constraint(): +# """GET /calculation/budget-constraint/{session_id} — require_any_role blocks Management +# Query params: cost_threshold (float, default 100) +# """ +# for user in USER_LIST: +# res = requests.get( +# f"{BASE_URL}/calculation/budget-constraint/{OVERHAUL_SESSION_UUID}?cost_threshold={BUDGET_THRESHOLD}", +# headers=user_auth_headers(user["authorization_token"]), +# ) +# expected = 200 if user["name"] in NON_MANAGEMENT else 403 +# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" diff --git a/src/testing/prod_integration_test.py b/src/testing/prod_integration_test.py new file mode 100644 index 0000000..e70cbf8 --- /dev/null +++ b/src/testing/prod_integration_test.py @@ -0,0 +1,285 @@ +""" +Integration tests for Optimum OH backend. +All endpoints are sourced from frontend API calls using OPTIMUM_OH_API_URL. + +Run with: + pytest src/testing/integration_test.py -v +""" + +import pytest +import requests + +BASE_URL = "http://100.125.115.116:8000/optimumoh" +AUTH_BASE_URL = "http://100.125.115.116:8000/auth" + +# Replace with a valid token before running +AUTH_TOKEN = "Bearer " + +HEADERS = { + "Authorization": AUTH_TOKEN, + "Content-Type": "application/json", +} + +# Placeholder IDs — replace with real values from your database before running +OVERHAUL_SESSION_ID = 1 +OVERHAUL_SCHEDULE_ID = 1 +OVERHAUL_EQUIPMENT_ID = 1 +SCOPE_JOB_ID = 1 +CALCULATION_ID = 1 +BUDGET_THRESHOLD = 100000 +RISK_COST = 1 +LOCATION_TAG = "LOCATION-001" +SCOPE_NAME = "scope-name-example" +OVERHAUL_SESSION_HISTORY_ID = 1 +ASSET_NUM = "ASSET-001" +SCOPE_EQUIPMENT_JOB_ID = 1 +EQUIPMENT_ACTIVITY_ID = 1 +EAF_INPUT = 0.85 +DURATION = 8760 +SCOPE_EQUIPMENT_ACTIVITY_ID = 1 +ITEMNUM = "ITEM-001" +SPREADSHEET_LINK = "https://docs.google.com/spreadsheets/d/example_spreadsheet_id/edit" +# UUID-format placeholder — replace with a real UUID from your database +OVERHAUL_SESSION_UUID = "00000000-0000-0000-0000-000000000001" + +# --------------------------------------------------------------------------- +# Users +# --------------------------------------------------------------------------- +USER_LIST = [ + {"id": 1, "name": "devmustbeadmin", "pass": "", "email": "devmustbeadmin@mail.com", "authorization_token": ""}, + # {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "devmustbeengineer@mail.com", "authorization_token": ""}, + # {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "devmustbeapplicationadmin@mail.com", "authorization_token": ""}, + # {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "devmustbemanagement@mail.com", "authorization_token": ""}, +] + +# --------------------------------------------------------------------------- +# Role-based access constants +# --------------------------------------------------------------------------- +# Endpoints guarded by require_any_role(*ALLOWED_ROLES) block Management entirely. +# Endpoints with only required_permission(READ) allow Management (has READ in ACL). +# Endpoints with no guard at all allow every role. +NON_MANAGEMENT = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin"} +ALL_ALLOWED = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin", "devmustbemanagement"} + + +def user_auth_headers(token: str) -> dict: + return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + +def get_csrf_token(path: str, auth_token: str) -> str: + """Fetch a one-time CSRF token from be-auth for the given endpoint path. + + Args: + path: The endpoint path as seen by be-optimumoh (e.g. "/spareparts"). + Do NOT include the "/optimumoh" Kong prefix. + auth_token: The raw JWT token string (without "Bearer " prefix). + + Returns: + The plain-text CSRF token to send in the X-CSRF-Token header. + """ + res = requests.post( + f"{AUTH_BASE_URL}/csrf-token", + headers={ + "Authorization": f"Bearer {auth_token}", + "Content-Type": "application/json", + "X-Target-Path": path, + }, + json={ + "_csrf_request": True, + } + ) + assert res.status_code == 200, f"Failed to get CSRF token for {path}: {res.status_code} {res.text}" + data = res.json() + return data["data"]["csrf_token"] + + +# ───────────────────────────────────────────── +# Overhaul Schedules +# ───────────────────────────────────────────── + +def test_get_overhaul_schedules(): + """GET /overhaul-schedules — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhaul-schedules", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + + +def test_get_overhaul_schedules_history(): + """GET /overhaul-schedules/history — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhaul-schedules/history", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + +# ───────────────────────────────────────────── +# Overhauls +# ───────────────────────────────────────────── + +def test_get_overhauls(): + """GET /overhauls — required_permission(READ) only; Management has READ in ACL""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhauls", headers=user_auth_headers(user["authorization_token"])) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# ───────────────────────────────────────────── +# Overhaul Gantt +# ───────────────────────────────────────────── + +def test_get_overhaul_gantt(): + """GET /overhaul-gantt — no role guard; all roles allowed""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhaul-gantt", headers=user_auth_headers(user["authorization_token"])) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +def test_get_overhaul_gantt_spreadsheet(): + """GET /overhaul-gantt/spreadsheet — no role guard; all roles allowed""" + for user in USER_LIST: + res = requests.get( + f"{BASE_URL}/overhaul-gantt/spreadsheet", + headers=user_auth_headers(user["authorization_token"]), + ) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + + +# ───────────────────────────────────────────── +# Spareparts +# ───────────────────────────────────────────── + +def test_get_spareparts(): + """GET /spareparts — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/spareparts", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + +def test_get_equipment_spareparts(): + """GET /equipment-spareparts/{location_tag} — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get( + f"{BASE_URL}/equipment-spareparts/{LOCATION_TAG}", + headers=user_auth_headers(user["authorization_token"]), + ) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + +def test_post_spareparts(): + """POST /spareparts — require_any_role blocks Management + Payload: itemnum (str), remark (str). CSRF token required. + """ + payload = { + "itemnum": ITEMNUM, + "remark": "Test remark for integration test", + } + for user in USER_LIST: + csrf_token = get_csrf_token("/spareparts", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post(f"{BASE_URL}/spareparts", json=payload, headers=headers) + if user["name"] in NON_MANAGEMENT: + assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + else: + assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" + +def test_post_overhaul_gantt_spreadsheet(): + """POST /overhaul-gantt/spreadsheet — no role guard; all roles allowed + Payload: spreadsheet_link (str). CSRF token required in production. + """ + payload = { + "spreadsheet_link": SPREADSHEET_LINK, + } + for user in USER_LIST: + csrf_token = get_csrf_token("/overhaul-gantt/spreadsheet", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/overhaul-gantt/spreadsheet", + json=payload, + headers=headers, + ) + assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" + +# ───────────────────────────────────────────── +# Scope Equipments +# ───────────────────────────────────────────── + +def test_get_scope_equipments(): + """GET /scope-equipments?page=1 — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/scope-equipments?page=1", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + + +def test_get_scope_equipments_available(): + """GET /scope-equipments/available/{scope_name} — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get( + f"{BASE_URL}/scope-equipments/available/{SCOPE_NAME}", + headers=user_auth_headers(user["authorization_token"]), + ) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}" + +# ───────────────────────────────────────────── +# Calculation — Time Constraint +# ───────────────────────────────────────────── + +def test_get_calculation_time_constraint_history(): + """GET /calculation/time-constraint — required_permission(READ) only; Management has READ""" + for user in USER_LIST: + res = requests.get( + f"{BASE_URL}/calculation/time-constraint", + headers=user_auth_headers(user["authorization_token"]), + ) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + +# ───────────────────────────────────────────── +# Overhaul Schedules (additional) +# ───────────────────────────────────────────── + +def test_get_overhaul_schedules(): + """GET /overhaul-schedules — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhaul-schedules", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + + +def test_get_overhaul_schedules_history(): + """GET /overhaul-schedules/history — require_any_role blocks Management""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhaul-schedules/history", headers=user_auth_headers(user["authorization_token"])) + expected = 200 if user["name"] in NON_MANAGEMENT else 403 + assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}" + +# ───────────────────────────────────────────── +# Overhauls sub-routes +# ───────────────────────────────────────────── + +def test_get_overhauls_system_components(): + """GET /overhauls/system-components — required_permission(READ); Management has READ""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/overhauls/system-components", headers=user_auth_headers(user["authorization_token"])) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" + + +# ───────────────────────────────────────────── +# Scope Equipments +# ───────────────────────────────────────────── + +def test_get_scope_equipments(): + """GET /scope-equipments — required_permission(READ) only; Management has READ in ACL""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/scope-equipments", headers=user_auth_headers(user["authorization_token"])) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}" + + +# ───────────────────────────────────────────── +# Workscopes +# ───────────────────────────────────────────── + +def test_get_workscopes(): + """GET /workscopes — no role guard; all roles allowed""" + for user in USER_LIST: + res = requests.get(f"{BASE_URL}/workscopes", headers=user_auth_headers(user["authorization_token"])) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}" \ No newline at end of file