Skip to content
Snippets Groups Projects
Select Git revision
  • develop default protected
  • master protected
  • feature/refacture-l3-core-services
  • feature/10GGBS-NAT-980
  • fix/NAT-1009/fix-redeploy-base-config-if-there-is-a-vprn
  • feature/NAT-732-ias-to-re-interconnect
  • 2.48
  • 2.47
  • 2.46
  • 2.45
  • 2.44
  • 2.43
  • 2.42
  • 2.41
  • 2.40
  • 2.39
  • 2.38
  • 2.37
  • 2.36
  • 2.35
  • 2.34
  • 2.33
  • 2.32
  • 2.31
  • 2.29
  • 2.28
26 results
Blame Permalink
lan_switch_interconnect_fixtures.py 4.75 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107 






from uuid import uuid4

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_blocks.lan_switch_interconnect import (
    LanSwitchInterconnectInterfaceBlockInactive,
    LanSwitchInterconnectRouterSideBlockInactive,
    LanSwitchInterconnectSwitchSideBlockInactive,
)
from gso.products.product_types.lan_switch_interconnect import (
    ImportedLanSwitchInterconnectInactive,
    LanSwitchInterconnectInactive,
)
from gso.products.product_types.router import Router
from gso.products.product_types.switch import Switch
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType


@pytest.fixture()
def lan_switch_interconnect_subscription_factory(
    faker, geant_partner, router_subscription_factory, switch_subscription_factory
):
    def create_subscription(
        description: str | None = None,
        partner: dict | None = None,
        status: SubscriptionLifecycle | None = None,
        start_date: str | None = "2024-01-01T10:20:30+01:02",
        lan_switch_interconnect_description: str | None = None,
        lan_switch_interconnect_ip_network: IPv4NetworkType | None = None,
        address_space: AddressSpace | None = None,
        minimum_links: int | None = None,
        router_side_node: UUIDstr | None = None,
        router_side_ae_iface: str | None = None,
        router_side_ae_members: list[dict[str, str]] | None = None,
        router_side_ipv4_address: IPv4AddressType | None = None,
        switch_side_switch: UUIDstr | None = None,
        switch_side_ae_iface: str | None = None,
        switch_side_ae_members: list[dict[str, str]] | None = None,
        switch_side_ipv4_address: IPv4AddressType | None = None,
        *,
        is_imported: bool | None = True,
    ) -> UUIDstr:
        if partner is None:
            partner = geant_partner
        if is_imported:
            product_id = get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT)
            subscription = LanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])
        else:
            product_id = get_product_id_by_name(ProductName.IMPORTED_LAN_SWITCH_INTERCONNECT)
            subscription = ImportedLanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])

        router_side_ae_members = router_side_ae_members or [
            LanSwitchInterconnectInterfaceBlockInactive.new(
                uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
            )
            for _ in range(2)
        ]
        switch_side_ae_members = switch_side_ae_members or [
            LanSwitchInterconnectInterfaceBlockInactive.new(
                uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
            )
            for _ in range(2)
        ]

        subscription.lan_switch_interconnect.lan_switch_interconnect_description = (
            lan_switch_interconnect_description or faker.sentence()
        )
        subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = (
            lan_switch_interconnect_ip_network or faker.ipv4_network()
        )
        subscription.lan_switch_interconnect.address_space = address_space or AddressSpace.PRIVATE
        subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
        subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
            uuid4(),
            node=router_side_node or Router.from_subscription(router_subscription_factory()).router,
            ae_iface=router_side_ae_iface or faker.network_interface(),
            ae_members=router_side_ae_members,
            ipv4_address=router_side_ipv4_address or faker.ipv4(),
        )
        subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
            uuid4(),
            switch=switch_side_switch or Switch.from_subscription(switch_subscription_factory()).switch,
            ae_iface=switch_side_ae_iface or faker.network_interface(),
            ae_members=switch_side_ae_members,
            ipv4_address=switch_side_ipv4_address or faker.ipv4(),
        )

        subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
        subscription.insync = True
        subscription.description = description or faker.sentence()
        subscription.start_date = start_date

        if status:
            subscription.status = status

        subscription.save()
        db.session.commit()

        return str(subscription.subscription_id)

    return create_subscription