ct/bmc_hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
bmc_hub/tests/test_subscription_network_provisioning.py
Christian 3311b8e590 Add comprehensive tests for internet connections module, invoice parsing, and subscription provisioning 
- Implement tests for the internet connections module, covering routes, IP range creation, and connection validation.
- Add tests for the Invoice2DataService to validate extraction from GlobalConnect invoices.
- Create tests for subscription network provisioning, ensuring proper handling of network items and IP allocations.
- Include validation checks for subtotal mismatches and ensure error handling for missing IP selections.
2026-07-09 23:44:30 +02:00

431 lines
17 KiB
Python
Raw Permalink Blame History

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"]
Reference in New Issue View Git Blame Copy Permalink