Newer
Older
import ipaddress
import pytest
from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products.product_blocks.router import RouterRole, RouterVendor
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.router import RouterInactive
from gso.products.product_types.site import Site, SiteInactive
from gso.services import subscriptions
CUSTOMER_ID: UUIDstr = "2f47f65a-0911-e511-80d0-005056956c1a"
@pytest.fixture
def site_subscription_factory(faker):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
site_name=None,
site_city=None,
site_country=None,
site_country_code=None,
site_latitude=None,
site_longitude=None,
site_bgp_community_id=None,
site_internal_id=None,
site_tier=SiteTier.TIER1,
site_ts_address=None,
) -> UUIDstr:
description = description or "Site Subscription"
site_name = site_name or faker.domain_word()
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
site_city = site_city or faker.city()
site_country = site_country or faker.country()
site_country_code = site_country_code or faker.country_code()
site_latitude = site_latitude or float(faker.latitude())
site_longitude = site_longitude or float(faker.longitude())
site_bgp_community_id = site_bgp_community_id or faker.pyint()
site_internal_id = site_internal_id or faker.pyint()
site_ts_address = site_ts_address or faker.ipv4()
product_id = subscriptions.get_product_id_by_name(ProductType.SITE)
site_subscription = SiteInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
site_subscription.site.site_city = site_city
site_subscription.site.site_name = site_name
site_subscription.site.site_country = site_country
site_subscription.site.site_country_code = site_country_code
site_subscription.site.site_latitude = site_latitude
site_subscription.site.site_longitude = site_longitude
site_subscription.site.site_bgp_community_id = site_bgp_community_id
site_subscription.site.site_internal_id = site_internal_id
site_subscription.site.site_tier = site_tier
site_subscription.site.site_ts_address = site_ts_address
site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
site_subscription.description = description
site_subscription.start_date = start_date
site_subscription.save()
db.session.commit()
return str(site_subscription.subscription_id)
return subscription_create
@pytest.fixture
def router_subscription_factory(site_subscription_factory, faker):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
router_fqdn=None,
router_ts_port=None,
router_access_via_ts=None,
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
router_si_ipv4_network=None,
router_ias_lt_ipv4_network=None,
router_ias_lt_ipv6_network=None,
router_vendor=RouterVendor.NOKIA,
router_role=RouterRole.PE,
router_site=None,
router_is_ias_connected=True,
) -> UUIDstr:
description = description or faker.text(max_nb_chars=30)
router_fqdn = router_fqdn or faker.domain_name(levels=4)
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
router_si_ipv4_network = router_si_ipv4_network or faker.ipv4_network()
router_ias_lt_ipv4_network = router_ias_lt_ipv4_network or faker.ipv4_network()
router_ias_lt_ipv6_network = router_ias_lt_ipv6_network or faker.ipv6_network()
router_site = router_site or site_subscription_factory()
product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
router_subscription = RouterInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
router_subscription.router.router_access_via_ts = router_access_via_ts
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
router_subscription.router.router_si_ipv4_network = router_si_ipv4_network
router_subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
router_subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
router_subscription.router.router_vendor = router_vendor
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
router_subscription.router.router_is_ias_connected = router_is_ias_connected
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.description = description
router_subscription.start_date = start_date
router_subscription.save()
db.session.commit()
return str(router_subscription.subscription_id)
return subscription_create