Skip to content
Snippets Groups Projects
Commit 8f77a86d authored by Erik Reid's avatar Erik Reid
Browse files

added migrations & commented-out sanity check

parent 76bc131a
No related branches found
No related tags found
No related merge requests found
......@@ -10,6 +10,10 @@ logger = logging.getLogger(__name__)
_SESSION_MAKER: Union[None, Session] = None
def mysql_dsn(username, password, hostname, db_name):
return f'mysql://{username}:{password}@{hostname}/{db_name}'
@contextlib.contextmanager
def session_scope(
callback_before_close: Optional[Callable] = None) -> Iterator[Session]:
......@@ -31,19 +35,13 @@ def session_scope(
session.close()
def init_db_model(username: str, password: str, hostname: str, db_name: str):
dsn = 'mysql://{user}:{pwd}@{host}/{db}'.format(
user=username,
pwd=password,
host=hostname,
db=db_name)
def init_db_model(dsn):
global _SESSION_MAKER
# cf. https://docs.sqlalchemy.org/en
# /latest/orm/extensions/automap.html
engine = create_engine(dsn, pool_size=10, max_overflow=0,
pool_recycle=3600)
engine = create_engine(
dsn, pool_size=10, max_overflow=0, pool_recycle=3600)
_SESSION_MAKER = sessionmaker(bind=engine)
"""create nodes table
Revision ID: 330fd3aba6ca
Revises:
Create Date: 2022-10-17 12:17:59.346758
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '330fd3aba6ca'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
'nodes',
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('fqdn', sa.String(256), nullable=False))
def downgrade() -> None:
op.drop_table('nodes')
"""create line card and port tables
Revision ID: 8901d3bca270
Revises: 330fd3aba6ca
Create Date: 2022-10-17 12:45:02.028717
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '8901d3bca270'
down_revision = '330fd3aba6ca'
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
'line_cards',
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('position', sa.Integer, nullable=False),
sa.Column('model', sa.String(256), nullable=False),
sa.Column('node_id', sa.Integer, sa.ForeignKey('nodes.id'), nullable=False))
op.create_table(
'ports',
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String(256), nullable=False),
sa.Column('line_card_id', sa.Integer, sa.ForeignKey('line_cards.id'), nullable=False))
def downgrade() -> None:
op.drop_table('ports')
op.drop_table('line_cards')
......@@ -2,21 +2,11 @@ import contextlib
from unittest.mock import patch
import pytest
from sqlalchemy import create_engine, func
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from resource_management.db import session_scope
class DBTestUtils(object):
@staticmethod
@contextlib.contextmanager
def session():
with session_scope() as session:
yield session
from resource_management.db import model
@pytest.fixture
def resources_db():
......@@ -33,4 +23,4 @@ def resources_db():
with patch(
'resource_management.db._SESSION_MAKER',
sessionmaker(bind=engine)):
yield DBTestUtils
yield # wait until caller context ends
from resource_management.db import model
from resource_management.db import session_scope
def test_sanity_check(resources_db):
with resources_db.session() as session:
with session_scope() as session:
node = model.Node(fqdn='a.b.c')
session.add(node)
......@@ -17,9 +18,23 @@ def test_sanity_check(resources_db):
session.add(port)
# new session - previous rows should have been committed
with resources_db.session() as session:
with session_scope() as session:
node = session.query(model.Node) \
.filter(model.Node.fqdn == 'a.b.c').one()
assert len(node.line_cards) == 10
for lc in node.line_cards:
assert len(lc.ports) == 20
# TODO: do this in a tmp container ...
# from resource_management.db import mysql_dsn
# from resource_management.migrations import migration_utils
#
#
# def test_migration():
# dsn = mysql_dsn(
# username='dummy',
# password='dummy-pass',
# hostname='localhost',
# db_name='resources_test')
# migration_utils.upgrade(dsn=dsn)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment