Skip to content
Snippets Groups Projects
Commit b7f43e60 authored by Neda Moeini's avatar Neda Moeini
Browse files

Created Opengear migration.

parent ca86e154
No related branches found
No related tags found
1 merge request!202Feature/nat 458 opengear domain model
Pipeline #86593 failed
Stage: tox
Stage: documentation
Stage: sonarqube
This commit is part of merge request !202. Comments created here will be created in the context of that merge request.
Hide whitespace changes
Inline Side-by-side
gso/migrations/versions/2024-04-26_32cad119b7c4_add_opengear_product.py 0 → 100644
+ 116
− 0
View file @ b7f43e60
"""Add Opengear product..
Revision ID: 32cad119b7c4
Revises: 393acfa175c0
Create Date: 2024-04-26 11:12:36.852353
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = '32cad119b7c4'
down_revision = '393acfa175c0'
branch_labels = None
depends_on = None
def upgrade() -> None:
conn = op.get_bind()
conn.execute(sa.text("""
INSERT INTO products (name, description, product_type, tag, status) VALUES ('Opengear', 'An Opengear', 'Opengear', 'OPENGEAR', 'active') RETURNING products.product_id
"""))
conn.execute(sa.text("""
INSERT INTO product_blocks (name, description, tag, status) VALUES ('OpengearBlock', 'An OpengearBlock', 'OPENGEAR_BLOCK', 'active') RETURNING product_blocks.product_block_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('opengear_wan_address', 'The WAN address of the Opengear device.') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('opengear_wan_netmask', 'The WAN netmask of the Opengear device.') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('opengear_wan_gateway', 'The WAN gateway of the Opengear device.') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('opengear_hostname', 'The hostname of the Opengear device.') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO product_product_blocks (product_id, product_block_id) VALUES ((SELECT products.product_id FROM products WHERE products.name IN ('Opengear')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock')))
"""))
conn.execute(sa.text("""
INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock')))
"""))
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 ('OpengearBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_hostname')))
"""))
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 ('OpengearBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_address')))
"""))
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 ('OpengearBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_netmask')))
"""))
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 ('OpengearBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_gateway')))
"""))
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 ('OpengearBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_hostname'))
"""))
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 ('OpengearBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_hostname'))
"""))
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 ('OpengearBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_address'))
"""))
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 ('OpengearBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_address'))
"""))
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 ('OpengearBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_netmask'))
"""))
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 ('OpengearBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_netmask'))
"""))
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 ('OpengearBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_gateway'))
"""))
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 ('OpengearBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_gateway'))
"""))
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 ('opengear_wan_address', 'opengear_wan_netmask', 'opengear_wan_gateway', 'opengear_hostname'))
"""))
conn.execute(sa.text("""
DELETE FROM resource_types WHERE resource_types.resource_type IN ('opengear_wan_address', 'opengear_wan_netmask', 'opengear_wan_gateway', 'opengear_hostname')
"""))
conn.execute(sa.text("""
DELETE FROM product_product_blocks WHERE product_product_blocks.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Opengear')) AND product_product_blocks.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock'))
"""))
conn.execute(sa.text("""
DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock'))
"""))
conn.execute(sa.text("""
DELETE FROM subscription_instances WHERE subscription_instances.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock'))
"""))
conn.execute(sa.text("""
DELETE FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock')
"""))
conn.execute(sa.text("""
DELETE FROM processes WHERE processes.pid IN (SELECT processes_subscriptions.pid FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Opengear'))))
"""))
conn.execute(sa.text("""
DELETE FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Opengear')))
"""))
conn.execute(sa.text("""
DELETE FROM subscription_instances WHERE subscription_instances.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Opengear')))
"""))
conn.execute(sa.text("""
DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Opengear'))
"""))
conn.execute(sa.text("""
DELETE FROM products WHERE products.name IN ('Opengear')
"""))
gso/products/__init__.py
+ 4
− 0
View file @ b7f43e60
......@@ -16,6 +16,7 @@ from gso.products.product_types.router import Router
from gso.products.product_types.site import Site
from gso.products.product_types.super_pop_switch import SuperPopSwitch
from gso.products.product_types.switch import Switch
from gso.products.product_types.opengear import Opengear
class ProductName(strEnum):
......@@ -29,6 +30,7 @@ class ProductName(strEnum):
SWITCH = "Switch"
LAN_SWITCH_INTERCONNECT = "LAN Switch Interconnect"
POP_VLAN = "Pop VLAN"
OPENGEAR = "Opengear"
class ProductType(strEnum):
......@@ -42,6 +44,7 @@ class ProductType(strEnum):
SWITCH = Switch.__name__
LAN_SWITCH_INTERCONNECT = LanSwitchInterconnect.__name__
POP_VLAN = PopVlan.__name__
OPENGEAR = Opengear.__name__
SUBSCRIPTION_MODEL_REGISTRY.update(
......@@ -54,5 +57,6 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
ProductName.SWITCH.value: Switch,
ProductName.LAN_SWITCH_INTERCONNECT.value: LanSwitchInterconnect,
ProductName.POP_VLAN.value: PopVlan,
ProductName.OPENGEAR.value: Opengear,
},
)
gso/products/product_blocks/opengear.py
+ 5
− 0
View file @ b7f43e60
......@@ -39,8 +39,13 @@ class OpengearBlockProvisioning(OpengearBlockInactive, lifecycle=[SubscriptionLi
class OpengearBlock(OpengearBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An Opengear that's currently deployed in the network."""
#: The hostname of the Opengear device.
opengear_hostname: str
#: The site where the Opengear device is located.
opengear_site: SiteBlock
#: The WAN address of the Opengear device.
opengear_wan_address: ipaddress.IPv4Address
#: The WAN netmask of the Opengear device.
opengear_wan_netmask: ipaddress.IPv4Address
#: The WAN gateway of the Opengear device.
opengear_wan_gateway: ipaddress.IPv4Address
gso/products/product_types/opengear.py
+ 1
− 1
View file @ b7f43e60
......@@ -18,7 +18,7 @@ class OpengearProvisioning(OpengearInactive, lifecycle=[SubscriptionLifecycle.PR
opengear: OpengearBlockProvisioning
class OpenGear(OpengearProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
class Opengear(OpengearProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An Opengear that is currently active."""
opengear: OpengearBlock
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment