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

initial model & sanity check test case

parent 9cd9c012
No related branches found
No related tags found
No related merge requests found
...@@ -3,3 +3,5 @@ requests ...@@ -3,3 +3,5 @@ requests
sqlalchemy sqlalchemy
alembic alembic
mysqlclient mysqlclient
pytest
\ No newline at end of file
import contextlib
import logging
from typing import Optional, Union, Callable, Iterator
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker, Session
logger = logging.getLogger(__name__)
_SESSION_MAKER: Union[None, Session] = None
@contextlib.contextmanager
def session_scope(
callback_before_close: Optional[Callable] = None) -> Iterator[Session]:
# best practice is to keep session scope separate from data processing
# cf. https://docs.sqlalchemy.org/en/13/orm/session_basics.html
assert _SESSION_MAKER
session = _SESSION_MAKER()
try:
yield session
session.commit()
if callback_before_close:
callback_before_close()
except SQLAlchemyError:
logger.exception("caught sql layer exception, rolling back")
session.rollback()
raise # re-raise, will be handled by main consumer
finally:
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)
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)
_SESSION_MAKER = sessionmaker(bind=engine)
"""
Resources Database
=========================
This database hold information about physical router resources
"""
import logging
from typing import Any
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
logger = logging.getLogger(__name__)
# https://github.com/python/mypy/issues/2477a
base_schema: Any = declarative_base()
class Node(base_schema):
__tablename__ = 'nodes'
id = Column('id', Integer, primary_key=True)
fqdn = Column('fqdn', String, nullable=False)
line_cards = relationship(
"LineCard",
back_populates="node",
cascade="all, delete, delete-orphan")
class LineCard(base_schema):
__tablename__ = 'line_cards'
id = Column('id', Integer, primary_key=True)
position = Column('position', Integer, nullable=False)
model = Column('model', String, nullable=False)
node_id = Column('node_id', Integer, ForeignKey('nodes.id'))
node = relationship('Node', back_populates='line_cards')
ports = relationship(
'Port',
back_populates='line_card',
cascade="all, delete, delete-orphan")
class Port(base_schema):
__tablename__ = 'ports'
id = Column('id', Integer, primary_key=True)
name = Column('name', String, nullable=False)
line_card_id = Column(
'line_card_id', Integer, ForeignKey('line_cards.id'))
line_card = relationship('LineCard', back_populates='ports')
import logging
import os
from alembic.config import Config
from alembic import command
logger = logging.getLogger(__name__)
DEFAULT_MIGRATIONS_DIRECTORY = os.path.dirname(__file__)
# def _mysql_dsn(db_username, db_password, db_hostname, db_name):
# return 'mysql://{username}:{password}@{hostname}/{dbname}'.format(
# username=db_username,
# password=db_password,
# hostname=db_hostname,
# dbname=db_name)
def upgrade(dsn, migrations_directory=DEFAULT_MIGRATIONS_DIRECTORY):
"""
migrate db to head version
cf. https://stackoverflow.com/a/43530495,
https://stackoverflow.com/a/54402853
:param dsn:
:param migrations_directory: full path to migrations directory
(default is this directory)
:return:
"""
logging.debug(f'running migrations against dsn {dsn}')
alembic_config = Config()
alembic_config.set_main_option('script_location', migrations_directory)
alembic_config.set_main_option('sqlalchemy.url', dsn)
command.upgrade(alembic_config, 'head')
import contextlib
from unittest.mock import patch
import pytest
from sqlalchemy import create_engine, func
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
@pytest.fixture
def resources_db():
# cf. https://stackoverflow.com/a/33057675
engine = create_engine(
'sqlite://',
connect_args={'check_same_thread': False},
poolclass=StaticPool,
echo=False)
model.base_schema.metadata.create_all(engine)
with patch(
'resource_management.db._SESSION_MAKER',
sessionmaker(bind=engine)):
yield DBTestUtils
from resource_management.db import model
def test_sanity_check(resources_db):
with resources_db.session() as session:
node = model.Node(fqdn='a.b.c')
session.add(node)
for position in range(10):
line_card = model.LineCard(
model='XYZ', position=position, node=node)
session.add(line_card)
for x in range(20):
port = model.Port(name=str(x), line_card=line_card)
session.add(port)
# new session - previous rows should have been committed
with resources_db.session() 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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment