CIzz22
/
be-optimumoh
1
0
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update: Integration test new

Browse Source
oh_security
TAMDeveloper13 2 months ago
parent 54fe0e1c12
commit 091bab1714
2 changed files with 448 additions and 587 deletions
Show Stats Download Patch File Download Diff File

676
src/testing/integration_test.py
Unescape Escape View File

@ -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

359
src/testing/prod_integration_test.py
Unescape Escape View File

@ -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 <your_token_here>"
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}"
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}"

Write Preview
Loading…
Cancel
Save