Commit d5fbfc4a authored by Miguel Catalan's avatar Miguel Catalan
Browse files

addding verbosity to the tests

parent fe7da8f9

tests/oran/test_create_get_delete_qod_session.py

+16 −4
Original line number Diff line number Diff line 
# -*- coding: utf-8 -*-

import time

from pprint import pformat



import pytest
@@ -41,28 +42,39 @@ def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient): 
    }



    # Create policy

    print("[Test] Creating QoD policy with payload:")

    print(pformat(camara_session))

    response = oran_client.create_qod_session(camara_session)

    print("[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")

    time.sleep(10)



    # Verify policy exists

    try:

        oran_client.get_qod_session(policy_id)

        get_resp = oran_client.get_qod_session(policy_id)

        print(f"[Test] GET policy {policy_id} 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}")

        oran_client.delete_qod_session(policy_id)

    except OranHttpError as e:

        pytest.fail(f"Failed to delete oran policy: {e}")



    # Wait 10 seconds for the automated expiration in NEF

    time.sleep(10)

    # Optional short wait to allow backend cleanup

    print("[Test] Sleeping 5s before verifying deletion")

    time.sleep(5)



    # Verify deletion (expect a failure on get)

    with pytest.raises(OranHttpError):

    print(f"[Test] Expecting OranHttpError on GET after deletion (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

+14 −3
Original line number Diff line number Diff line 
# -*- coding: utf-8 -*-

import time

from pprint import pformat



import pytest
@@ -43,20 +44,30 @@ def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient): 
    }



    # Create policy

    print("[Test] Creating QoD policy with payload:")

    print(pformat(camara_session))

    response = oran_client.create_qod_session(camara_session)

    print("[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:

        oran_client.get_qod_session(policy_id)

        get_resp = oran_client.get_qod_session(policy_id)

        print(f"[Test] GET after create (policy_id={policy_id}):")

        print(pformat(get_resp))

    except OranHttpError as e:

        pytest.fail(f"Policy should exist right after creation: {e}")



    # Wait slightly longer than the duration to ensure expiry

    buffer_seconds = 5

    time.sleep(duration_seconds + buffer_seconds)

    wait_secs = duration_seconds + buffer_seconds

    print(f"[Test] Waiting for expiry: sleeping {wait_secs}s")

    time.sleep(wait_secs)



    # After expiry, the policy should not exist anymore (expect OranHttpError/404)

    with pytest.raises(OranHttpError):

    print(f"[Test] Expecting OranHttpError on GET after expiry (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

+56 −13
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ import json 
import threading

import time

from http.server import BaseHTTPRequestHandler, HTTPServer

from pprint import pformat

from socketserver import ThreadingMixIn

from typing import List
@@ -27,14 +28,20 @@ def _make_handler(storage: List[dict]): 
                payload = json.loads(body.decode("utf-8") or "{}")

            except Exception:

                payload = {"raw": body.decode("utf-8", errors="ignore")}

            storage.append(

                {

            record = {

                "path": self.path,

                "headers": dict(self.headers),

                "payload": payload,

                "ts": time.time(),

            }

            )

            storage.append(record)

            # Verbose output of received callback

            try:

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

            except Exception:

                print("[Notify] Received callback at", record["path"])  # best-effort

            self.send_response(204)

            self.end_headers()
@@ -49,8 +56,7 @@ def _make_handler(storage: List[dict]): 
def notification_server():

    """Spin up a tiny HTTP server to capture ORAN NEF callbacks.



    Binds to 0.0.0.0:40000 so the callback URL is http://localhost:40000/callback

    matching existing examples in tests.

    Binds to the host used in test_cases; callback URL is printed for visibility.

    """

    host, port = "192.168.40.50", 40000

    received: List[dict] = []
@@ -63,6 +69,7 @@ def notification_server(): 
        "url": f"http://{host}:{port}/callback",

        "received": received,

    }

    print(f"[Notify] Callback server listening at {server['url']}")

    try:

        yield server

    finally:
@@ -87,6 +94,7 @@ def _wait_for_callbacks(store: List[dict], min_new: int, timeout: float, start_l 
    deadline = time.time() + timeout

    while time.time() < deadline:

        if len(store) - start_len >= min_new:

            print(f"[Notify] Callback(s) arrived: new={len(store) - start_len}, total={len(store)}")

            return True

        time.sleep(0.2)

    return False
@@ -111,8 +119,12 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_ 
        "notificationDestination": notification_server["url"],

    }



    print("[Test] Creating policy with payload:")

    print(pformat(camara_session))

    start_len = len(notification_server["received"])

    response = oran_client.create_qod_session(camara_session)

    print("[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"
@@ -120,11 +132,18 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_ 
    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:")

        for rec in notification_server["received"][start_len:]:

            print(pformat(rec))



    # Wait until after expiry

    time.sleep(duration_seconds + 5)

    wait_secs = duration_seconds + 5

    print(f"[Test] Waiting for expiry: sleep {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})")

    with pytest.raises(OranHttpError):

        oran_client.get_qod_session(policy_id)
@@ -132,6 +151,10 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_ 
    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:")

        for rec in notification_server["received"][start_len:]:

            print(pformat(rec))





@pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True)
@@ -152,8 +175,12 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n 
        "notificationDestination": notification_server["url"],

    }



    print("[Test] Creating policy with payload:")

    print(pformat(camara_session))

    start_len = len(notification_server["received"])

    response = oran_client.create_qod_session(camara_session)

    print("[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"
@@ -161,24 +188,40 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n 
    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:")

        for rec in notification_server["received"][start_len:]:

            print(pformat(rec))



    # Wait 10 seconds and verify exists

    print("[Test] Sleeping 10s before GET")

    time.sleep(10)

    print(f"[Test] GET policy {policy_id}")

    try:

        oran_client.get_qod_session(policy_id)

        get_resp = oran_client.get_qod_session(policy_id)

        print("[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}")

    oran_client.delete_qod_session(policy_id)



    time.sleep(15)

    # Give time for deletion callback to arrive

    print("[Test] Waiting 10s for deletion callback")

    time.sleep(10)



    # Expect an additional callback (e.g., deletion)

    assert _wait_for_callbacks(

        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:")

        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})")

    with pytest.raises(OranHttpError):

        oran_client.get_qod_session(policy_id)