-
Karel van Klink authoredKarel van Klink authored
opengear_fixtures.py 2.77 KiB
import pytest
from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_types.opengear import ImportedOpengearInactive, OpengearInactive
from gso.products.product_types.site import Site
from gso.services import subscriptions
@pytest.fixture()
def opengear_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
opengear_site=None,
opengear_hostname=None,
opengear_wan_address=None,
opengear_wan_netmask=None,
opengear_wan_gateway=None,
status: SubscriptionLifecycle | None = None,
partner: dict | None = None,
*,
is_imported: bool | None = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
description = description or faker.text(max_nb_chars=30)
opengear_site = opengear_site or site_subscription_factory()
opengear_hostname = opengear_hostname or faker.domain_name(levels=4)
opengear_wan_address = opengear_wan_address or faker.ipv4()
opengear_wan_netmask = opengear_wan_netmask or faker.ipv4()
opengear_wan_gateway = opengear_wan_gateway or faker.ipv4()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.OPENGEAR)
opengear_subscription = OpengearInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OPENGEAR)
opengear_subscription = ImportedOpengearInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
opengear_subscription.opengear.opengear_site = Site.from_subscription(opengear_site).site
opengear_subscription.opengear.opengear_hostname = opengear_hostname
opengear_subscription.opengear.opengear_wan_address = opengear_wan_address
opengear_subscription.opengear.opengear_wan_netmask = opengear_wan_netmask
opengear_subscription.opengear.opengear_wan_gateway = opengear_wan_gateway
opengear_subscription = SubscriptionModel.from_other_lifecycle(
opengear_subscription, SubscriptionLifecycle.ACTIVE
)
opengear_subscription.description = description
opengear_subscription.start_date = start_date
if status:
opengear_subscription.status = status
opengear_subscription.save()
db.session.commit()
return str(opengear_subscription.subscription_id)
return subscription_create