Add new attribute to site

gso/migrations/versions/2024-12-04_fc7bd696014e_add_optical_equipment_attribute_to_site.py

+"""Add optical equipment attribute to Site.
+
+Revision ID: fc7bd696014e
+Revises: 79a76b22ca53
+Create Date: 2024-12-04 10:15:40.529552
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'fc7bd696014e'
+down_revision = '79a76b22ca53'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('site_contains_optical_equipment', 'Whether a site contains optical equipment') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('site_contains_optical_equipment')))
+    """))
+    conn.execute(sa.text("""
+WITH rt_id AS (SELECT resource_type_id FROM resource_types WHERE resource_type = 'site_contains_optical_equipment') INSERT INTO subscription_instance_values (subscription_instance_id, resource_type_id, value) SELECT subscription_instance_id, rt_id.resource_type_id, 'True' FROM rt_id, subscription_instances WHERE product_block_id = (SELECT product_block_id FROM product_blocks WHERE name = 'SiteBlock');
+    """))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('site_contains_optical_equipment'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('site_contains_optical_equipment'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('site_contains_optical_equipment'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('site_contains_optical_equipment')
+    """))

gso/oss-params-example.json

@@ -54,7 +54,7 @@
     },
     "LAN_SWITCH_INTERCONNECT": {
       "V4": {"containers":  ["10.2.0.0/16"], "networks":  [], "mask": 24},
-      "V6": {"containers":  [], "networks": [], "mask": 126},
+      "V6": {"containers":  ["beef:cafe::/56"], "networks": [], "mask": 64},
       "domain_name": ".geant.net",
       "dns_view": "default",
       "network_view": "default"

gso/products/product_blocks/site.py

@@ -38,6 +38,7 @@ class SiteBlockInactive(
     site_bgp_community_id: int | None = None
     site_tier: SiteTier | None = None
     site_ts_address: IPAddress | None = None
+    site_contains_optical_equipment: bool | None = None
 
 
 class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -53,6 +54,7 @@ class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.
     site_bgp_community_id: int
     site_tier: SiteTier
     site_ts_address: IPAddress
+    site_contains_optical_equipment: bool
 
 
 class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -82,3 +84,5 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
     #:  access. This is required in case a link goes down, or when a router is initially added to the network and it
     #:  does not have any IP trunks connected to it.
     site_ts_address: IPAddress
+    #:  Whether this site contains optical equipment, which dictates the need for a DCN management VLAN
+    site_contains_optical_equipment: bool

gso/utils/types/base_site.py

@@ -24,3 +24,4 @@ class BaseSiteValidatorModel(BaseModel):
     site_latitude: LatitudeCoordinate
     site_longitude: LongitudeCoordinate
     partner: str
+    site_contains_optical_equipment: bool = True

gso/workflows/site/create_imported_site.py

@@ -55,6 +55,7 @@ def initialize_subscription(
     site_internal_id: int,
     site_ts_address: IPAddress,
     site_tier: SiteTier,
+    site_contains_optical_equipment: bool,  # noqa: FBT001
 ) -> State:
     """Initialise the subscription object with all input."""
     subscription.site.site_name = site_name
@@ -67,6 +68,7 @@ def initialize_subscription(
     subscription.site.site_internal_id = site_internal_id
     subscription.site.site_tier = site_tier
     subscription.site.site_ts_address = site_ts_address
+    subscription.site.site_contains_optical_equipment = site_contains_optical_equipment
 
     subscription.description = f"Site in {site_city}, {site_country}"
 

gso/workflows/site/create_site.py

@@ -85,6 +85,7 @@ def initialize_subscription(
     subscription.site.site_internal_id = site_internal_id
     subscription.site.site_tier = site_tier
     subscription.site.site_ts_address = site_ts_address
+    subscription.site.site_contains_optical_equipment = True
 
     subscription.description = f"Site in {site_city}, {site_country}"
 

gso/workflows/switch/create_switch.py

@@ -40,7 +40,7 @@ def _initial_input_form_generator(product_name: str) -> FormGenerator:
         hostname: str
         ts_port: PortNumber
         vendor: ReadOnlyField(Vendor.JUNIPER, default_type=Vendor)  # type: ignore[valid-type]
-        model: SwitchModel = Choice("Switch model", SwitchModel.values())
+        model: SwitchModel = Choice("Switch model", SwitchModel.values())  # type: ignore[assignment, arg-type]
 
         @model_validator(mode="after")
         def hostname_must_be_available(self) -> Self: