From 091bab17147b32a32662257600f375868942aa01 Mon Sep 17 00:00:00 2001 From: TAMDeveloper13 Date: Thu, 21 May 2026 10:20:01 +0000 Subject: [PATCH] Update: Integration test new --- src/testing/integration_test.py | 676 ++++++--------------------- src/testing/prod_integration_test.py | 359 +++++++++++--- 2 files changed, 448 insertions(+), 587 deletions(-) diff --git a/src/testing/integration_test.py b/src/testing/integration_test.py index 8f4dca4..e9dd15e 100644 --- a/src/testing/integration_test.py +++ b/src/testing/integration_test.py @@ -21,35 +21,33 @@ HEADERS = { } # 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 +OVERHAUL_SESSION_ID = "3704c82c-cfc5-47f7-813a-1661f89b0738" +# OVERHAUL_EQUIPMENT_ID = 1 ## Source code got commented +SCOPE_JOB_ID = "db590764-72cb-4eb1-a30f-014116df0472" +CALCULATION_ID = "44f483f3-bfe4-4094-a59f-b97a10f2fea6" 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 +LOCATION_TAG = "3AL-F501A" +SCOPE_NAME = "A" +# OVERHAUL_SESSION_HISTORY_ID = 1 +ASSET_NUM = "A19825" +# SCOPE_EQUIPMENT_JOB_ID = 1 ## Source code got commented +# EQUIPMENT_ACTIVITY_ID = 1 ## Source code got commented EAF_INPUT = 0.85 -DURATION = 8760 -SCOPE_EQUIPMENT_ACTIVITY_ID = 1 -ITEMNUM = "ITEM-001" +DURATION = 17520 +SCOPE_EQUIPMENT_ACTIVITY_ID = "4527eb2b-0bc4-46b3-80a1-e1d2f67882f8" +ITEMNUM = "206510" 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": ""}, + {"id": 1, "name": "devmustbeadmin", "pass": "", "email": "", "authorization_token": ""}, + {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "", "authorization_token": ""}, + {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "", "authorization_token": ""}, + {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "", "authorization_token": ""}, ] # --------------------------------------------------------------------------- @@ -95,11 +93,25 @@ def get_csrf_token(path: str, auth_token: str) -> str: return data["data"]["csrf_token"] + + + +# OH GANTT + +# # [[BUG IN THIS ROUTE]] +# 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}" + + + # ───────────────────────────────────────────── # Overhaul Session # ───────────────────────────────────────────── -# # Note: Source code got commented +# # [[Source code got commented]] # def test_get_overhaul_session(): # """GET /overhaul-session?all=1 — require_any_role blocks Management""" # for user in USER_LIST: @@ -108,8 +120,7 @@ def get_csrf_token(path: str, auth_token: str) -> str: # 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 +# # [[Source code got commented]] # def test_post_overhaul_session(): # """POST /overhaul-session — require_any_role + CREATE blocks Management""" # payload = { @@ -118,21 +129,20 @@ def get_csrf_token(path: str, auth_token: str) -> str: # "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"])) +# csrf_token = get_csrf_token("/overhaul-session", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post(f"{BASE_URL}/overhaul-session", json=payload, headers=headers) # 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 # # ───────────────────────────────────────────── - +# # [[Attribute error]] # def test_post_overhaul_schedules(): # """POST /overhaul-schedules — require_any_role + CREATE blocks Management""" # payload = { @@ -143,7 +153,10 @@ def get_csrf_token(path: str, auth_token: str) -> str: # "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"Testing payload: {payload}, headers: {user_auth_headers(user['authorization_token'])}") + # csrf_token = get_csrf_token("/overhaul-schedules", user["authorization_token"]) + # headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + # res = requests.post(f"{BASE_URL}/overhaul-schedules", json=payload, headers=headers) # 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}" @@ -151,48 +164,15 @@ def get_csrf_token(path: str, auth_token: str) -> str: # assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" +# # [[Source code got commented]] # 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, -# } +# """POST /overhaul-schedules/delete/{scope_id} — require_any_role + DELETE blocks Management. CSRF required.""" # for user in USER_LIST: +# csrf_token = get_csrf_token("/overhaul-schedules/delete", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # 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"]), +# f"{BASE_URL}/overhaul-schedules/delete/{OVERHAUL_SCHEDULE_ID}", +# headers=headers, # ) # 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}" @@ -200,21 +180,25 @@ def get_csrf_token(path: str, auth_token: str) -> str: # assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" -# ───────────────────────────────────────────── -# Overhaul Jobs -# ───────────────────────────────────────────── +# # ───────────────────────────────────────────── +# # Overhaul Jobs +# # ───────────────────────────────────────────── + +# # [[Source code got commented]] # def test_post_overhaul_jobs(): -# """POST /overhaul-jobs/{scope_job_id} — no role guard; all roles allowed""" +# """POST /overhaul-jobs/{overhaul_equipment_id} — no role guard; all roles allowed""" # payload = { -# "overhaul_equipment_id": OVERHAUL_EQUIPMENT_ID, +# "job_ids": [], # } # for user in USER_LIST: +# csrf_token = get_csrf_token("/overhaul-jobs", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # res = requests.post( # f"{BASE_URL}/overhaul-jobs/{SCOPE_JOB_ID}", # json=payload, -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" @@ -223,7 +207,7 @@ def get_csrf_token(path: str, auth_token: str) -> str: # Scope Equipments # ───────────────────────────────────────────── - +# # [[Database error: Unconsumed column names: scope_overhaul, assetnum, type, removal_date]] # def test_post_scope_equipments(): # """POST /scope-equipments — require_any_role blocks Management""" # payload = { @@ -231,14 +215,16 @@ def get_csrf_token(path: str, auth_token: str) -> str: # "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"])) +# csrf_token = get_csrf_token("/scope-equipments", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post(f"{BASE_URL}/scope-equipments", json=payload, headers=headers) # 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}" - +# # [[No endpoint exists]] # def test_delete_scope_equipment(): # """DELETE /scope-equipments/delete/{id} — require_any_role blocks Management""" # for user in USER_LIST: @@ -252,102 +238,78 @@ def get_csrf_token(path: str, auth_token: str) -> str: # assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}" -# ───────────────────────────────────────────── -# Scope Equipment Jobs -# ───────────────────────────────────────────── +# # ───────────────────────────────────────────── +# # Scope Equipment Jobs +# # ───────────────────────────────────────────── +# # [[Source code got commented]] # def test_post_scope_equipment_jobs(): # """POST /scope-equipment-jobs/{assetnum} — no role guard; all roles allowed""" # payload = { -# "job_id": SCOPE_JOB_ID, +# "job_ids": [], # } # for user in USER_LIST: +# csrf_token = get_csrf_token("/scope-equipment-jobs", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # res = requests.post( # f"{BASE_URL}/scope-equipment-jobs/{ASSET_NUM}", # json=payload, -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" - +# # [[Source code got commented]] # def test_delete_scope_equipment_job(): -# """DELETE /scope-equipment-jobs/delete/{id} — no role guard; all roles allowed""" +# """POST /equipment-workscopes/delete/{scope_job_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"]), +# csrf_token = get_csrf_token("/equipment-workscopes/delete", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post( +# f"{BASE_URL}/equipment-workscopes/delete/{SCOPE_EQUIPMENT_JOB_ID}", +# headers=headers, # ) # assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" - # # ───────────────────────────────────────────── -# # Calculation — Budget Constraint +# # Overhaul Jobs # # ───────────────────────────────────────────── -# 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 -# } +# # [[Source code got commented]] +# def test_get_overhaul_jobs(): +# """GET /overhaul-jobs/{overhaul_equipment_id} — no role guard; all roles allowed""" # for user in USER_LIST: -# res = requests.post( -# f"{BASE_URL}/calculation/time-constraint", -# json=payload, +# res = requests.get( +# f"{BASE_URL}/overhaul-jobs/{OVERHAUL_EQUIPMENT_ID}", # 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}" +# assert res.status_code == 200, f"[{user['name']}] expected 200, 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}" - +# # ───────────────────────────────────────────── +# # Calculation — Budget Constraint +# # ───────────────────────────────────────────── -# def test_put_calculation_time_constraint_update(): -# """PUT /calculation/time-constraint/update/{calculationId} — required_permission(EDIT); Management has no EDIT""" +# # [[AttributeError]] +# def test_post_calculation_time_constraint(): +# """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE""" # payload = { -# "risk_cost": RISK_COST, +# "ohSessionId": OVERHAUL_SESSION_ID, +# "costPerFailure": 0, +# "overhaulCost": 0, +# } +# params = { +# "with_results": 0, # } # for user in USER_LIST: -# res = requests.put( -# f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}", +# csrf_token = get_csrf_token("/calculation/time-constraint", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post( +# f"{BASE_URL}/calculation/time-constraint", # json=payload, -# headers=user_auth_headers(user["authorization_token"]), +# params=params, +# headers=headers, # ) # 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}" @@ -357,38 +319,36 @@ def get_csrf_token(path: str, auth_token: str) -> str: # # 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}" - - +# # [[Source code got commented]] # def test_post_equipment_activities(): -# """POST /equipment-activities — no role guard; all roles allowed""" +# """POST /equipment-activities — no role guard; all roles allowed +# Note: route commented out in api.py; active equivalent is POST /overhaul-activity/{session_id} +# """ # payload = { -# "location_tag": LOCATION_TAG, -# "workscope": "Inspection", +# "location_tags": [LOCATION_TAG], # } # for user in USER_LIST: +# csrf_token = get_csrf_token("/equipment-activities", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # res = requests.post( # f"{BASE_URL}/equipment-activities", # json=payload, -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}" +# # [[Source code got commented]] # def test_delete_equipment_activity(): -# """DELETE /equipment-activities/delete/{id} — no role guard; all roles allowed""" +# """POST /equipment-activities/delete/{id} — no role guard; all roles allowed +# Note: route commented out in api.py; active equivalent is POST /overhaul-activity/delete/{session}/{location_tag} +# """ # for user in USER_LIST: -# res = requests.delete( +# csrf_token = get_csrf_token("/equipment-activities/delete", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post( # f"{BASE_URL}/equipment-activities/delete/{EQUIPMENT_ACTIVITY_ID}", -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}" @@ -397,96 +357,19 @@ def get_csrf_token(path: str, auth_token: str) -> str: # # Jobs # # ───────────────────────────────────────────── +# # [[Source code got commented]] # 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 -# # ───────────────────────────────────────────── +# ───────────────────────────────────────────── +# Overhauls sub-routes +# ───────────────────────────────────────────── +# # [[TypeError]] # def test_get_overhauls_schedules(): # """GET /overhauls/schedules — required_permission(READ); Management has READ""" # for user in USER_LIST: @@ -494,7 +377,7 @@ def get_csrf_token(path: str, auth_token: str) -> str: # 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}" - +# # [[TypeError]] # def test_get_overhauls_critical_parts(): # """GET /overhauls/critical-parts — required_permission(READ); Management has READ""" # for user in USER_LIST: @@ -503,109 +386,27 @@ def get_csrf_token(path: str, auth_token: str) -> str: # 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}" - - +# # [[Attribute Error]] # 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}", +# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}/{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 # # ───────────────────────────────────────────── +# # [[TypeError]] # def test_post_workscopes(): # """POST /workscopes — no role guard; all roles allowed # Payload: workscope (str) @@ -621,154 +422,38 @@ def get_csrf_token(path: str, auth_token: str) -> str: # # "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"])) +# csrf_token = get_csrf_token("/workscopes", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} +# res = requests.post(f"{BASE_URL}/workscopes", json=payload, headers=headers) # 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}" - +# ───────────────────────────────────────────── +# Equipment Workscopes +# ───────────────────────────────────────────── +# # [[NameError]] # 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: +# csrf_token = get_csrf_token("/equipment-workscopes/delete", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # res = requests.post( # f"{BASE_URL}/equipment-workscopes/delete/{SCOPE_JOB_ID}", -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # 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 # # ───────────────────────────────────────────── +# # [[AttributeError]] # def test_post_calculation_time_constraint(): # """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE # Payload: overhaulCost (float), ohSessionId (UUID), costPerFailure (float) @@ -776,14 +461,16 @@ def get_csrf_token(path: str, auth_token: str) -> str: # """ # payload = { # "overhaulCost": 0, -# "ohSessionId": OVERHAUL_SESSION_UUID, +# "ohSessionId": OVERHAUL_SESSION_ID, # "costPerFailure": 0, # } # for user in USER_LIST: +# csrf_token = get_csrf_token("/calculation/time-constraint", user["authorization_token"]) +# headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} # res = requests.post( # f"{BASE_URL}/calculation/time-constraint", # json=payload, -# headers=user_auth_headers(user["authorization_token"]), +# headers=headers, # ) # print(f"[{user['name']}], got {res.status_code}, response: {res.text}") # if user["name"] in NON_MANAGEMENT: @@ -791,17 +478,7 @@ def get_csrf_token(path: str, auth_token: str) -> str: # 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}" - - +# # [[AttributeError]] # def test_get_calculation_time_constraint_parameters(): # """GET /calculation/time-constraint/parameters — required_permission(READ); Management has READ # Query params: calculation_id (optional) @@ -813,19 +490,7 @@ def get_csrf_token(path: str, auth_token: str) -> str: # ) # 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}" - - +# # [[AttributeError]] # 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: @@ -836,69 +501,20 @@ def get_csrf_token(path: str, auth_token: str) -> str: # 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 # # ───────────────────────────────────────────── +# # [[AttributeError]] # 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}", +# f"{BASE_URL}/calculation/budget-constraint/{OVERHAUL_SESSION_ID}?cost_threshold={BUDGET_THRESHOLD}", # headers=user_auth_headers(user["authorization_token"]), # ) # expected = 200 if user["name"] in NON_MANAGEMENT else 403 diff --git a/src/testing/prod_integration_test.py b/src/testing/prod_integration_test.py index e70cbf8..421bec8 100644 --- a/src/testing/prod_integration_test.py +++ b/src/testing/prod_integration_test.py @@ -12,44 +12,28 @@ 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 +OVERHAUL_SESSION_ID = "3704c82c-cfc5-47f7-813a-1661f89b0738" +SCOPE_JOB_ID = "db590764-72cb-4eb1-a30f-014116df0472" +CALCULATION_ID = "44f483f3-bfe4-4094-a59f-b97a10f2fea6" 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 +LOCATION_TAG = "3AL-F501A" +SCOPE_NAME = "A" +ASSET_NUM = "A19825" EAF_INPUT = 0.85 -DURATION = 8760 -SCOPE_EQUIPMENT_ACTIVITY_ID = 1 -ITEMNUM = "ITEM-001" +DURATION = 17520 +SCOPE_EQUIPMENT_ACTIVITY_ID = "4527eb2b-0bc4-46b3-80a1-e1d2f67882f8" +ITEMNUM = "206510" 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": ""}, + {"id": 1, "name": "devmustbeadmin", "pass": "", "email": "", "authorization_token": ""}, + {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "", "authorization_token": ""}, + {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "", "authorization_token": ""}, + {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "", "authorization_token": ""}, ] # --------------------------------------------------------------------------- @@ -88,8 +72,10 @@ def get_csrf_token(path: str, auth_token: str) -> str: "_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"] @@ -104,6 +90,38 @@ def test_get_overhaul_schedules(): 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_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: + csrf_token = get_csrf_token("/overhaul-schedules/update", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/overhaul-schedules/update/{OVERHAUL_SESSION_ID}", + json=payload, + headers=headers, + ) + 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_get_overhaul_schedules_history(): """GET /overhaul-schedules/history — require_any_role blocks Management""" @@ -112,6 +130,23 @@ def test_get_overhaul_schedules_history(): 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_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: + csrf_token = get_csrf_token(f"/overhaul-schedules/delete/{OVERHAUL_SESSION_ID}", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/overhaul-schedules/delete/{OVERHAUL_SESSION_ID}", + headers=headers, + ) + 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}" + # ───────────────────────────────────────────── # Overhauls # ───────────────────────────────────────────── @@ -126,12 +161,6 @@ def test_get_overhauls(): # 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: @@ -220,18 +249,6 @@ def test_get_scope_equipments_available(): 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) @@ -264,22 +281,250 @@ def test_get_overhauls_system_components(): # ───────────────────────────────────────────── -# Scope Equipments +# Workscopes # ───────────────────────────────────────────── -def test_get_scope_equipments(): - """GET /scope-equipments — required_permission(READ) only; Management has READ in ACL""" +def test_get_workscopes(): + """GET /workscopes — no role guard; all roles allowed""" 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}" + 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}" +def test_get_workscopes_by_id(): + """GET /workscopes/{scope_equipment_activity_id}?activity_id=... — no role guard; all roles allowed + """ + for user in USER_LIST: + res = requests.get( + f"{BASE_URL}/workscopes/{SCOPE_EQUIPMENT_ACTIVITY_ID}", + params={"activity_id": 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: + csrf_token = get_csrf_token("/workscopes/update", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/workscopes/update/{SCOPE_EQUIPMENT_ACTIVITY_ID}", + json=payload, + params={"activity_id": SCOPE_EQUIPMENT_ACTIVITY_ID}, + headers=headers, + ) + 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: + csrf_token = get_csrf_token("/workscopes/delete", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/workscopes/delete/{SCOPE_EQUIPMENT_ACTIVITY_ID}", + params={"activity_id": SCOPE_EQUIPMENT_ACTIVITY_ID}, + headers=headers, + ) + 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}" # ───────────────────────────────────────────── -# Workscopes +# Overhaul Activity # ───────────────────────────────────────────── -def test_get_workscopes(): - """GET /workscopes — no role guard; all roles allowed""" +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}/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 + 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}" + +def test_post_overhaul_activity_current(): + """POST /overhaul-activity/{session_id} — require_any_role blocks Management + Payload: location_tags (List[str]). CSRF token required. + """ + payload = { + "location_tags": [LOCATION_TAG], + } + for user in USER_LIST: + csrf_token = get_csrf_token("/overhaul-activity", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}", + 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_delete_overhaul_activity(): + """POST /overhaul-activity/delete/{overhaul_session}/{location_tag} — require_any_role blocks Management. CSRF required.""" + for user in USER_LIST: + csrf_token = get_csrf_token(f"/overhaul-activity/delete/{OVERHAUL_SESSION_ID}/{LOCATION_TAG}", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/overhaul-activity/delete/{OVERHAUL_SESSION_ID}/{LOCATION_TAG}", + headers=headers, + ) + 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 Equipments History +# ───────────────────────────────────────────── + +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_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}" + + +# ───────────────────────────────────────────── +# 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: + csrf_token = get_csrf_token("/equipment-workscopes", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/equipment-workscopes/{ASSET_NUM}", + json=payload, + headers=headers, + ) + 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}" + + +# ───────────────────────────────────────────── +# 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_ID}&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 — Time Constraint +# ───────────────────────────────────────────── + +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_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_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: + csrf_token = get_csrf_token("/calculation/time-constraint/simulation", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}/simulation", + json=payload, + headers=headers, + ) + 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 { location_tag (str), name (str), is_included (bool) }. CSRF token required in production. + """ + payload = [ + {"location_tag": LOCATION_TAG, "name": "Test Equipment", "is_included": True}, + ] + for user in USER_LIST: + csrf_token = get_csrf_token(f"/calculation/time-constraint/update/{CALCULATION_ID}", user["authorization_token"]) + headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token} + res = requests.post( + f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}", + json=payload, + headers=headers, + ) + 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}" + + +# ───────────────────────────────────────────── +# 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"]), + ) + assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"