import sys import asyncio from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from fastapi.testclient import TestClient from main import app def test_subscription_by_sag_includes_network_provisioning(monkeypatch): from app.subscriptions.backend import router as subscription_router def fake_execute_query_single(query, params=None): if "FROM sag_subscriptions s" in query and "WHERE s.sag_id = %s" in query: return { "id": 44, "subscription_number": "SUB-44", "sag_id": 9, "customer_id": 1662, "customer_name": "BMC Networks", "product_name": "BMCnet 100/100", "billing_interval": "monthly", "billing_day": 1, "price": 499, "start_date": "2026-07-01", "end_date": None, "status": "draft", "notes": None, } if "FROM internet_connections_connections" in query and "subscription_id = %s" in query: return None return None def fake_execute_query(query, params=None): if "FROM sag_subscription_items i" in query: return [ { "id": 100, "line_no": 1, "product_id": 501, "product_name": "BMCnet 100/100", "product_type": "service", "attributes_json": {"network": {"kind": "internet_access", "download_mbps": 100, "upload_mbps": 100}}, "description": "BMCnet 100/100", "quantity": 1, "unit_price": 399, "line_total": 399, "period_from": None, "period_to": None, "requires_serial_number": False, "serial_number": None, "billing_blocked": False, "billing_block_reason": None, }, { "id": 101, "line_no": 2, "product_id": 502, "product_name": "/30 IP", "product_type": "service", "attributes_json": {"network": {"kind": "ip_allocation", "ip_prefix_length": 30}}, "description": "/30 IP", "quantity": 1, "unit_price": 100, "line_total": 100, "period_from": None, "period_to": None, "requires_serial_number": False, "serial_number": None, "billing_blocked": False, "billing_block_reason": None, }, ] return [] monkeypatch.setattr(subscription_router, "execute_query_single", fake_execute_query_single) monkeypatch.setattr(subscription_router, "execute_query", fake_execute_query) payload = asyncio.run(subscription_router.get_subscription_by_sag(9, True)) assert payload["requires_network_provisioning"] is True assert payload["network_provisioning"]["internet_items"][0]["download_mbps"] == 100 assert payload["network_provisioning"]["ip_items"][0]["ip_prefix_length"] == 30 def test_subscription_provisioning_endpoint_returns_shared_heads(monkeypatch): from app.modules.internet_connections.backend import router as internet_router def fake_execute_query_single(query, params=None): if "FROM sag_subscriptions s" in query: return { "id": 55, "subscription_number": "SUB-55", "sag_id": 7, "customer_id": 21, "customer_name": "Eksempel Kunde", "product_name": "BMCnet 100/100", "billing_interval": "monthly", "price": 699, "status": "draft", "start_date": "2026-07-01", } return None def fake_execute_query(query, params=None): if "FROM sag_subscription_items i" in query: return [ { "id": 201, "line_no": 1, "product_id": 601, "product_name": "BMCnet 100/100", "product_type": "service", "attributes_json": {"network": {"kind": "internet_access", "download_mbps": 100, "upload_mbps": 100}}, "description": "BMCnet 100/100", "quantity": 1, "unit_price": 499, "line_total": 499, }, { "id": 202, "line_no": 2, "product_id": 602, "product_name": "/30 IP", "product_type": "service", "attributes_json": {"network": {"kind": "ip_allocation", "ip_prefix_length": 30}}, "description": "/30 IP", "quantity": 1, "unit_price": 200, "line_total": 200, }, ] if "ic.subscription_id = %s" in query: return [] if "ic.allocation_model = 'shared'" in query: return [ { "id": 88, "parent_id": None, "name": "BMC Networks · Albertslund", "provider": "GlobalConnect", "customer_id": 1662, "customer_name": "BMC Networks", "parent_name": None, "address": "Rydagervej 27, 2620 Albertslund", "status": "active", "monthly_cost": 0, "sales_price": 0, "margin_amount": 0, "technology": "Fiber", "connection_type": "fiber", "circuit_number": "NKA020900", "speed_mbps": 1000, "upload_mbps": 1000, "download_mbps": 1000, "monitoring_url": None, "contract_start": None, "contract_end": None, "allocation_model": "shared", "value_type": "bmc_networks", "value_label": None, "subscription_id": None, "subscription_number": None, "subscription_product_name": None, "subscription_customer_name": None, "ip_range_count": 1, "total_ip_addresses": 2, "in_use_ip_addresses": 0, "reserved_ip_addresses": 0, "available_ip_addresses": 2, } ] if "FROM internet_connections_ip_ranges ir" in query and "GROUP BY ir.id, c.name" in query: return [ { "id": 301, "connection_id": 88, "name": "87.116.1.200/30", "cidr": "87.116.1.200/30", "description": None, "provider_reference": "NKA020900", "contract_number": None, "customer_id": None, "service_address": "Rydagervej 27, 2620 Albertslund", "monthly_cost": 0, "sales_price": 0, "customer_name": None, "total_addresses": 2, "available_addresses": 2, "reserved_addresses": 0, "in_use_addresses": 0, } ] return [] monkeypatch.setattr(internet_router, "execute_query_single", fake_execute_query_single) monkeypatch.setattr(internet_router, "execute_query", fake_execute_query) client = TestClient(app) response = client.get("/api/v1/internet-connections/subscriptions/55/provisioning") assert response.status_code == 200 payload = response.json() assert payload["network_provisioning"]["requires_provisioning"] is True assert payload["shared_heads"][0]["address"] == "Rydagervej 27, 2620 Albertslund" assert payload["shared_heads"][0]["available_matching_ranges"][0]["cidr"] == "87.116.1.200/30" def test_subscription_provisioning_endpoint_can_offer_derived_subrange(monkeypatch): from app.modules.internet_connections.backend import router as internet_router def fake_execute_query_single(query, params=None): if "FROM sag_subscriptions s" in query: return { "id": 56, "subscription_number": "SUB-56", "sag_id": 7, "customer_id": 21, "customer_name": "Eksempel Kunde", "product_name": "BMCnet 500/500 + /28", "billing_interval": "monthly", "price": 699, "status": "draft", "start_date": "2026-07-01", } return None def fake_execute_query(query, params=None): if "FROM sag_subscription_items i" in query: return [ { "id": 301, "line_no": 1, "product_id": 701, "product_name": "BMCnet 500/500", "product_type": "service", "attributes_json": {"network": {"kind": "internet_access", "download_mbps": 500, "upload_mbps": 500}}, "description": "BMCnet 500/500", "quantity": 1, "unit_price": 499, "line_total": 499, }, { "id": 302, "line_no": 2, "product_id": 702, "product_name": "/28 IP", "product_type": "service", "attributes_json": {"network": {"kind": "ip_allocation", "ip_prefix_length": 28}}, "description": "/28 IP", "quantity": 1, "unit_price": 200, "line_total": 200, }, ] if "ic.subscription_id = %s" in query: return [] if "ic.allocation_model = 'shared'" in query: return [ { "id": 188, "parent_id": None, "name": "BMC Networks · Albertslund", "provider": "GlobalConnect", "customer_id": 1662, "customer_name": "BMC Networks", "parent_name": None, "address": "Rydagervej 27, 2620 Albertslund", "status": "active", "monthly_cost": 0, "sales_price": 0, "margin_amount": 0, "technology": "Fiber", "connection_type": "fiber", "circuit_number": "NKA020900", "speed_mbps": 1000, "upload_mbps": 1000, "download_mbps": 1000, "monitoring_url": None, "contract_start": None, "contract_end": None, "allocation_model": "shared", "value_type": "bmc_networks", "value_label": None, "subscription_id": None, "subscription_number": None, "subscription_product_name": None, "subscription_customer_name": None, "ip_range_count": 1, "total_ip_addresses": 62, "in_use_ip_addresses": 0, "reserved_ip_addresses": 0, "available_ip_addresses": 62, } ] if "FROM internet_connections_ip_ranges ir" in query and "GROUP BY ir.id, c.name" in query: return [ { "id": 401, "connection_id": 188, "name": "83.136.94.128/26", "cidr": "83.136.94.128/26", "description": None, "provider_reference": "NKA021047", "contract_number": None, "customer_id": None, "service_address": "Herstedvang 14, 2620 Albertslund", "monthly_cost": 0, "sales_price": 0, "customer_name": None, "total_addresses": 62, "available_addresses": 62, "reserved_addresses": 0, "in_use_addresses": 0, } ] return [] monkeypatch.setattr(internet_router, "execute_query_single", fake_execute_query_single) monkeypatch.setattr(internet_router, "execute_query", fake_execute_query) client = TestClient(app) response = client.get("/api/v1/internet-connections/subscriptions/56/provisioning") assert response.status_code == 200 payload = response.json() assert payload["shared_heads"][0]["available_matching_range_count"] == 4 assert payload["shared_heads"][0]["available_matching_ranges"][0]["requested_cidr"].endswith("/28") assert payload["shared_heads"][0]["available_matching_ranges"][0]["is_derived_candidate"] is True def test_subscription_provisioning_requires_ip_selection_for_each_ip_product(monkeypatch): from app.modules.internet_connections.backend import router as internet_router def fake_execute_query_single(query, params=None): if "FROM sag_subscriptions s" in query: return { "id": 66, "subscription_number": "SUB-66", "sag_id": 8, "customer_id": 77, "customer_name": "Kunde", "product_name": "BMCnet 100/100", "billing_interval": "monthly", "price": 799, "status": "draft", "start_date": "2026-07-01", } if "AND ic.id = %s AND ic.allocation_model = 'shared'" in query: return { "id": 90, "parent_id": None, "name": "BMC Networks · Shared", "provider": "GlobalConnect", "customer_id": 1662, "customer_name": "BMC Networks", "parent_name": None, "address": "Adresse 1", "status": "active", "monthly_cost": 0, "sales_price": 0, "margin_amount": 0, "technology": "Fiber", "connection_type": "fiber", "circuit_number": "NKA1", "speed_mbps": 1000, "upload_mbps": 1000, "download_mbps": 1000, "monitoring_url": None, "contract_start": None, "contract_end": None, "allocation_model": "shared", "value_type": "bmc_networks", "value_label": None, "subscription_id": None, "subscription_number": None, "subscription_product_name": None, "subscription_customer_name": None, "ip_range_count": 1, "total_ip_addresses": 2, "in_use_ip_addresses": 0, "reserved_ip_addresses": 0, "available_ip_addresses": 2, } return None def fake_execute_query(query, params=None): if "FROM sag_subscription_items i" in query: return [ { "id": 210, "line_no": 1, "product_id": 610, "product_name": "BMCnet 100/100", "product_type": "service", "attributes_json": {"network": {"kind": "internet_access", "download_mbps": 100, "upload_mbps": 100}}, "description": "BMCnet 100/100", "quantity": 1, "unit_price": 499, "line_total": 499, }, { "id": 211, "line_no": 2, "product_id": 611, "product_name": "/30 IP", "product_type": "service", "attributes_json": {"network": {"kind": "ip_allocation", "ip_prefix_length": 30}}, "description": "/30 IP", "quantity": 1, "unit_price": 300, "line_total": 300, }, ] if "ic.subscription_id = %s" in query: return [] return [] monkeypatch.setattr(internet_router, "execute_query_single", fake_execute_query_single) monkeypatch.setattr(internet_router, "execute_query", fake_execute_query) client = TestClient(app) response = client.post( "/api/v1/internet-connections/subscriptions/66/provision", json={"shared_connection_id": 90, "internet_item_id": 210, "ip_allocations": []}, ) assert response.status_code == 409 assert "IP-range" in response.json()["detail"]