Skip to content
Snippets Groups Projects

Clean up the repo a bit, and add some unit tests

Merged Karel van Klink requested to merge feature/add-unit-tests into develop
All threads resolved!
15 files
+ 279
231
Compare changes
  • Side-by-side
  • Inline
Files
15
"""Modify IP trunk model.
Revision ID: 394dc60d5c02
Revises: 01e42c100448
Create Date: 2023-10-11 17:55:38.289125
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = '394dc60d5c02'
down_revision = '01e42c100448'
branch_labels = None
depends_on = None
def upgrade() -> 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 ('IptrunkSideBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('iptrunk_side_ae_members_description'))
"""))
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 ('IptrunkSideBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('iptrunk_side_ae_members_description'))
"""))
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 ('IptrunkSideBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('iptrunk_side_ae_members'))
"""))
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 ('IptrunkSideBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('iptrunk_side_ae_members'))
"""))
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 ('iptrunk_side_ae_members_description', 'iptrunk_side_ae_members'))
"""))
conn.execute(sa.text("""
DELETE FROM resource_types WHERE resource_types.resource_type IN ('iptrunk_side_ae_members_description', 'iptrunk_side_ae_members')
"""))
conn.execute(sa.text("""
INSERT INTO product_blocks (name, description, tag, status) VALUES ('IptrunkInterfaceBlock', 'Interface in a LAG as part of an IP trunk', 'IPTINT', 'active') RETURNING product_blocks.product_block_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('interface_description', 'Description of a LAG interface') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('interface_name', 'Interface name of a LAG member') RETURNING resource_types.resource_type_id
"""))
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 ('IptrunkSideBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('IptrunkInterfaceBlock')))
"""))
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 ('IptrunkInterfaceBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_description')))
"""))
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 ('IptrunkInterfaceBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_name')))
"""))
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 ('IptrunkInterfaceBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_description'))
"""))
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 ('IptrunkInterfaceBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_description'))
"""))
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 ('IptrunkInterfaceBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_name'))
"""))
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 ('IptrunkInterfaceBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('interface_name'))
"""))
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 ('interface_description', 'interface_name'))
"""))
conn.execute(sa.text("""
DELETE FROM resource_types WHERE resource_types.resource_type IN ('interface_description', 'interface_name')
"""))
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 ('IptrunkSideBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('IptrunkInterfaceBlock'))
"""))
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 ('IptrunkInterfaceBlock'))
"""))
conn.execute(sa.text("""
DELETE FROM product_blocks WHERE product_blocks.name IN ('IptrunkInterfaceBlock')
"""))
Loading