addding verbosity to the tests (d6701197) · Commits · OOP / code / transformation-function-sdk

tests/oran/test_create_get_delete_qod_session.py

+14 −7

Original line number	Diff line number	Diff line
		@@ -42,39 +42,46 @@ def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient):
		}

		# Create policy
		print("[Test] Creating QoD policy with payload:")
		print("\n===== [Test] CREATE QoD policy =====")
		print("[Test] Payload:")
		print(pformat(camara_session))
		response = oran_client.create_qod_session(camara_session)
		print("[Test] Create response:")
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"

		# Wait 10 seconds
		print("[Test] Sleeping 10s before GET")
		print("\n===== [Test] WAIT before GET =====")
		print("[Test] Sleeping 10s")
		time.sleep(10)

		# Verify policy exists
		try:
		print("\n===== [Test] GET policy =====")
		print(f"[Test] policy_id={policy_id}")
		get_resp = oran_client.get_qod_session(policy_id)
		print(f"[Test] GET policy {policy_id} response:")
		print("[Test] GET response:")
		print(pformat(get_resp))
		except OranHttpError as e:
		pytest.fail(f"Policy should exist before deletion: {e}")

		# Delete policy
		try:
		print(f"[Test] DELETE policy {policy_id}")
		print("\n===== [Test] DELETE policy =====")
		print(f"[Test] policy_id={policy_id}")
		oran_client.delete_qod_session(policy_id)
		except OranHttpError as e:
		pytest.fail(f"Failed to delete oran policy: {e}")

		# Optional short wait to allow backend cleanup
		print("[Test] Sleeping 5s before verifying deletion")
		print("\n===== [Test] WAIT after DELETE =====")
		print("[Test] Sleeping 5s")
		time.sleep(5)

		# Verify deletion (expect a failure on get)
		print(f"[Test] Expecting OranHttpError on GET after deletion (policy_id={policy_id})")
		print("\n===== [Test] GET after DELETE (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		with pytest.raises(OranHttpError) as excinfo:
		oran_client.get_qod_session(policy_id)
		print(f"[Test] Received expected error: {excinfo.value}")

tests/oran/test_create_qod_session_with_expire.py

+9 −5

Original line number	Diff line number	Diff line
		@@ -44,18 +44,20 @@ def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient):
		}

		# Create policy
		print("[Test] Creating QoD policy with payload:")
		print("\n===== [Test] CREATE QoD policy =====")
		print("[Test] Payload:")
		print(pformat(camara_session))
		response = oran_client.create_qod_session(camara_session)
		print("[Test] Create response:")
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"

		# Immediately check it exists
		try:
		print("\n===== [Test] GET after create =====")
		get_resp = oran_client.get_qod_session(policy_id)
		print(f"[Test] GET after create (policy_id={policy_id}):")
		print(f"[Test] policy_id={policy_id}")
		print(pformat(get_resp))
		except OranHttpError as e:
		pytest.fail(f"Policy should exist right after creation: {e}")
		@@ -63,11 +65,13 @@ def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient):
		# Wait slightly longer than the duration to ensure expiry
		buffer_seconds = 5
		wait_secs = duration_seconds + buffer_seconds
		print(f"[Test] Waiting for expiry: sleeping {wait_secs}s")
		print("\n===== [Test] WAIT for expiry =====")
		print(f"[Test] Sleeping {wait_secs}s")
		time.sleep(wait_secs)

		# After expiry, the policy should not exist anymore (expect OranHttpError/404)
		print(f"[Test] Expecting OranHttpError on GET after expiry (policy_id={policy_id})")
		print("\n===== [Test] GET after expiry (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		with pytest.raises(OranHttpError) as excinfo:
		oran_client.get_qod_session(policy_id)
		print(f"[Test] Received expected error: {excinfo.value}")

tests/oran/test_qod_notifications.py

+31 −17

Original line number	Diff line number	Diff line
		@@ -37,6 +37,7 @@ def _make_handler(storage: List[dict]):
		storage.append(record)
		# Verbose output of received callback
		try:
		print("\n----- [Notify] Incoming Callback -----")
		print("[Notify] Received POST", record["path"]) # status set below
		print("[Notify] Headers:", pformat(record["headers"]))
		print("[Notify] Payload:", json.dumps(record["payload"], ensure_ascii=False))
		@@ -69,7 +70,8 @@ def notification_server():
		"url": f"http://{host}:{port}/callback",
		"received": received,
		}
		print(f"[Notify] Callback server listening at {server['url']}")
		print("\n===== [Notify] SERVER START =====")
		print(f"[Notify] Listening at {server['url']}")
		try:
		yield server
		finally:
		@@ -119,40 +121,45 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_
		"notificationDestination": notification_server["url"],
		}

		print("[Test] Creating policy with payload:")
		print("\n===== [Test] CREATE QoD policy (with notifications) =====")
		print("[Test] Payload:")
		print(pformat(camara_session))
		start_len = len(notification_server["received"])
		response = oran_client.create_qod_session(camara_session)
		print("[Test] Create response:")
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"

		# Expect at least one callback shortly after creation
		print("\n===== [Test] WAIT for creation callback =====")
		assert _wait_for_callbacks(
		notification_server["received"], 1, timeout=10, start_len=start_len
		), "Did not receive any callback after creation"
		if len(notification_server["received"]) > start_len:
		print("[Test] New callback(s) after create:")
		print("\n----- [Test] New callback(s) after create -----")
		for rec in notification_server["received"][start_len:]:
		print(pformat(rec))

		# Wait until after expiry
		wait_secs = duration_seconds + 5
		print(f"[Test] Waiting for expiry: sleep {wait_secs}s")
		print("\n===== [Test] WAIT for expiry =====")
		print(f"[Test] Sleeping {wait_secs}s")
		time.sleep(wait_secs)

		# After expiry, policy should be gone
		print(f"[Test] Expecting OranHttpError on GET after expiry (policy_id={policy_id})")
		print("\n===== [Test] GET after expiry (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		with pytest.raises(OranHttpError):
		oran_client.get_qod_session(policy_id)

		# Expect at least one additional callback (e.g., expiry)
		print("\n===== [Test] WAIT for expiry callback =====")
		assert _wait_for_callbacks(
		notification_server["received"], 2, timeout=10, start_len=start_len
		), "Did not receive post-expiry callback"
		if len(notification_server["received"]) > start_len:
		print("[Test] Callbacks collected:")
		print("\n----- [Test] Callbacks collected -----")
		for rec in notification_server["received"][start_len:]:
		print(pformat(rec))

		@@ -175,41 +182,47 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n
		"notificationDestination": notification_server["url"],
		}

		print("[Test] Creating policy with payload:")
		print("\n===== [Test] CREATE QoD policy (with notifications) =====")
		print("[Test] Payload:")
		print(pformat(camara_session))
		start_len = len(notification_server["received"])
		response = oran_client.create_qod_session(camara_session)
		print("[Test] Create response:")
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"

		# Expect at least one callback shortly after creation
		print("\n===== [Test] WAIT for creation callback =====")
		assert _wait_for_callbacks(
		notification_server["received"], 1, timeout=10, start_len=start_len
		), "Did not receive any callback after creation"
		if len(notification_server["received"]) > start_len:
		print("[Test] New callback(s) after create:")
		print("\n----- [Test] New callback(s) after create -----")
		for rec in notification_server["received"][start_len:]:
		print(pformat(rec))

		# Wait 10 seconds and verify exists
		print("[Test] Sleeping 10s before GET")
		print("\n===== [Test] WAIT before GET =====")
		print("[Test] Sleeping 10s")
		time.sleep(10)
		print(f"[Test] GET policy {policy_id}")
		print("\n===== [Test] GET policy =====")
		print(f"[Test] policy_id={policy_id}")
		try:
		get_resp = oran_client.get_qod_session(policy_id)
		print("[Test] GET response:")
		print("\n----- [Test] GET response -----")
		print(pformat(get_resp))
		except OranHttpError as e:
		pytest.fail(f"Policy should exist before deletion: {e}")

		# Delete policy
		print(f"[Test] DELETE policy {policy_id}")
		print("\n===== [Test] DELETE policy =====")
		print(f"[Test] policy_id={policy_id}")
		oran_client.delete_qod_session(policy_id)

		# Give time for deletion callback to arrive
		print("[Test] Waiting 10s for deletion callback")
		print("\n===== [Test] WAIT for deletion callback =====")
		print("[Test] Sleeping 10s")
		time.sleep(10)

		# Expect an additional callback (e.g., deletion)
		@@ -217,11 +230,12 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n
		notification_server["received"], 2, timeout=10, start_len=start_len
		), "Did not receive post-deletion callback"
		if len(notification_server["received"]) > start_len:
		print("[Test] Callbacks collected:")
		print("\n----- [Test] Callbacks collected -----")
		for rec in notification_server["received"][start_len:]:
		print(pformat(rec))

		# Verify policy is gone
		print(f"[Test] Expecting OranHttpError on GET after deletion (policy_id={policy_id})")
		print("\n===== [Test] GET after DELETE (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		with pytest.raises(OranHttpError):
		oran_client.get_qod_session(policy_id)

Admin message