Skip to content
Snippets Groups Projects
Select Git revision
  • feature/fix-vrf-modification
  • 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.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
  • 2.27
  • 2.26
  • 2.25
  • 2.24
27 results
Blame Permalink
network.py 5.09 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130 






"""API endpoints for network related operations."""

import ipaddress
from uuid import UUID

from fastapi import APIRouter
from orchestrator.domain import SubscriptionModel
from orchestrator.schemas.base import OrchestratorBaseModel
from orchestrator.services.subscriptions import build_extended_domain_model
from starlette import status

from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
from gso.services.subscriptions import get_active_iptrunk_subscriptions
from gso.utils.shared_enums import Vendor

router = APIRouter(prefix="/networks", tags=["Network"])


class SiteBlock(OrchestratorBaseModel):
    """Site block schema."""

    site_name: str
    site_city: str
    site_country: str
    site_latitude: LatitudeCoordinate
    site_longitude: LongitudeCoordinate
    site_internal_id: int


class RouterBlock(OrchestratorBaseModel):
    """Router block schema."""

    subscription_instance_id: UUID
    router_fqdn: str
    router_access_via_ts: bool
    router_lo_ipv4_address: ipaddress.IPv4Address
    router_lo_ipv6_address: ipaddress.IPv6Address
    router_lo_iso_address: str
    router_role: RouterRole
    vendor: Vendor
    router_site: SiteBlock


class IptrunkSideBlock(OrchestratorBaseModel):
    """Iptrunk side block schema."""

    subscription_instance_id: UUID
    iptrunk_side_node: RouterBlock


class IptrunkBlock(OrchestratorBaseModel):
    """Iptrunk block schema."""

    subscription_instance_id: UUID
    iptrunk_speed: PhysicalPortCapacity
    iptrunk_type: IptrunkType
    iptrunk_ipv4_network: ipaddress.IPv4Network
    iptrunk_ipv6_network: ipaddress.IPv6Network
    iptrunk_capacity: str
    iptrunk_isis_metric: int
    iptrunk_sides: list[IptrunkSideBlock]


class IptrunkSchema(OrchestratorBaseModel):
    """Iptrunk schema."""

    subscription_id: UUID
    insync: bool
    iptrunk: IptrunkBlock


class NetworkTopologyDomainModelSchema(OrchestratorBaseModel):
    """Network topology domain model schema."""

    iptrunks: list[IptrunkSchema]


def _calculate_iptrunk_capacity(iptrunk_sides: list, iptrunk_speed: PhysicalPortCapacity) -> str:
    """Calculate the total capacity of an IP trunk."""
    int_iptrunk_speed = int(iptrunk_speed.value.replace("G", ""))
    capacity = int_iptrunk_speed * len(iptrunk_sides[0]["iptrunk_side_ae_members"])
    return f"{capacity}G"


@router.get("/topology", status_code=status.HTTP_200_OK, response_model=NetworkTopologyDomainModelSchema)
def network_topology() -> NetworkTopologyDomainModelSchema:
    """Retrieve the network topology."""
    topology: dict = {"iptrunks": []}
    active_iptrunks = get_active_iptrunk_subscriptions()
    for iptrunk in active_iptrunks:
        subscription = SubscriptionModel.from_subscription(iptrunk["subscription_id"])
        extended_model = build_extended_domain_model(subscription)
        formatted_model = {
            "subscription_id": extended_model["subscription_id"],
            "insync": extended_model["insync"],
            "iptrunk": {
                "subscription_instance_id": extended_model["iptrunk"]["subscription_instance_id"],
                "iptrunk_speed": extended_model["iptrunk"]["iptrunk_speed"],
                "iptrunk_type": extended_model["iptrunk"]["iptrunk_type"],
                "iptrunk_ipv4_network": extended_model["iptrunk"]["iptrunk_ipv4_network"],
                "iptrunk_ipv6_network": extended_model["iptrunk"]["iptrunk_ipv6_network"],
                "iptrunk_isis_metric": extended_model["iptrunk"]["iptrunk_isis_metric"],
                "iptrunk_capacity": _calculate_iptrunk_capacity(
                    extended_model["iptrunk"]["iptrunk_sides"], extended_model["iptrunk"]["iptrunk_speed"]
                ),
                "iptrunk_sides": [
                    {
                        "subscription_instance_id": side["subscription_instance_id"],
                        "iptrunk_side_node": {
                            "subscription_instance_id": side["iptrunk_side_node"]["subscription_instance_id"],
                            "router_fqdn": side["iptrunk_side_node"]["router_fqdn"],
                            "router_access_via_ts": side["iptrunk_side_node"]["router_access_via_ts"],
                            "router_lo_ipv4_address": side["iptrunk_side_node"]["router_lo_ipv4_address"],
                            "router_lo_ipv6_address": side["iptrunk_side_node"]["router_lo_ipv6_address"],
                            "router_lo_iso_address": side["iptrunk_side_node"]["router_lo_iso_address"],
                            "router_role": side["iptrunk_side_node"]["router_role"],
                            "vendor": side["iptrunk_side_node"]["vendor"],
                            "router_site": side["iptrunk_side_node"]["router_site"],
                        },
                    }
                    for side in extended_model["iptrunk"]["iptrunk_sides"]
                ],
            },
        }
        topology["iptrunks"].append(IptrunkSchema(**formatted_model))

    return NetworkTopologyDomainModelSchema(**topology)