diff --git a/tests/test_smb_database.py b/tests/test_smb_database.py index 52a530a7..9372e85d 100644 --- a/tests/test_smb_database.py +++ b/tests/test_smb_database.py @@ -4,205 +4,149 @@ import asyncio import os from time import sleep -import pytest -import pytest_asyncio -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker, scoped_session - -from cme.cmedb import create_workspace, delete_workspace +from cme.cmedb import CMEDBMenu, create_workspace, create_db_engine, delete_workspace from cme.first_run import first_run_setup from cme.loaders.protocol_loader import protocol_loader from cme.logger import setup_logger, CMEAdapter -from cme.paths import WS_PATH -from sqlalchemy.dialects.sqlite import Insert +from cme.paths import CONFIG_PATH, WS_PATH -@pytest.fixture(scope="session") -def event_loop(): - return asyncio.get_event_loop() +class TestSmbDatabase: + def setup_class(self): + proto = "smb" + setup_logger() + logger = CMEAdapter() + first_run_setup(logger) + p_loader = protocol_loader() + protocols = p_loader.get_protocols() + create_workspace("test", p_loader, protocols) + protocol_db_path = p_loader.get_protocols()[proto]["dbpath"] + protocol_db_object = getattr(p_loader.load_protocol(protocol_db_path), "database") + db_path = os.path.join(WS_PATH, "test/smb.db") + db_engine = create_db_engine(db_path) + self.db = protocol_db_object(db_engine) -@pytest_asyncio.fixture(scope="session") -def db_engine(): - db_path = os.path.join(WS_PATH, "test/smb.db") - db_engine = create_engine( - f"sqlite:///{db_path}", - isolation_level="AUTOCOMMIT", - future=True - ) - yield db_engine - db_engine.dispose() + def teardown_class(self): + asyncio.run(self.db.shutdown_db()) + delete_workspace("test") + def test_add_host(self): + print(self.db) + self.db.add_host( + "127.0.0.1", + "localhost", + "TEST.DEV" + "Windows Testing 2023", + True, + False, + True, + False, + True + ) -@pytest_asyncio.fixture(scope="session") -async def db(db_engine): - proto = "smb" - setup_logger() - logger = CMEAdapter() - first_run_setup(logger) - p_loader = protocol_loader() - protocols = p_loader.get_protocols() - create_workspace("test", p_loader, protocols) + def test_update_host(self): + pass - protocol_db_path = p_loader.get_protocols()[proto]["dbpath"] - protocol_db_object = getattr(p_loader.load_protocol(protocol_db_path), "database") + def test_add_credential(self): + pass - db = protocol_db_object(db_engine) - yield db - db.shutdown_db() - delete_workspace("test") + def test_update_credential(self): + pass + def test_remove_credential(self): + pass -@pytest_asyncio.fixture(scope="session") -def sess(db_engine): - session_factory = sessionmaker( - bind=db_engine, - expire_on_commit=True - ) + def test_add_admin_user(self): + pass - Session = scoped_session( - session_factory - ) - sess = Session() - yield sess - sess.close() + def test_get_admin_relations(self): + pass + def test_remove_admin_relation(self): + pass -@pytest.mark.asyncio -async def test_add_host(db): - await db.add_host( - "127.0.0.1", - "localhost", - "TEST.DEV", - "Windows Testing 2023", - False, - True, - True, - True, - False, - False - ) + def test_is_credential_valid(self): + pass + def test_get_credentials(self): + pass -def test_update_host(db, sess): - host = { - "ip": "127.0.0.1", - "hostname": "localhost", - "domain": "TEST.DEV", - "os": "Windows Testing 2023", - "dc": False, - "smbv1": True, - "signing": True, - "spooler": True, - "zerologon": False, - "petitpotam": False - } - iq = Insert(db.HostsTable) - sess.execute(iq, [host]) + def test_get_credential(self): + pass + def test_is_credential_local(self): + pass -def test_add_credential(): - pass + def test_is_host_valid(self): + pass + def test_get_hosts(self): + pass -def test_update_credential(): - pass + def test_is_group_valid(self): + pass + def test_add_group(self): + pass -def test_remove_credential(): - pass + def test_get_groups(self): + pass + def test_get_group_relations(self): + pass -def test_add_admin_user(): - pass + def test_remove_group_relations(self): + pass -def test_get_admin_relations(): - pass + def test_is_user_valid(self): + pass -def test_remove_admin_relation(): - pass + def test_get_users(self): + pass -def test_is_credential_valid(): - pass + def test_get_user(self): + pass -def test_get_credentials(): - pass + def test_get_domain_controllers(self): + pass -def test_get_credential(): - pass + def test_is_share_valid(self): + pass -def test_is_credential_local(): - pass + def test_add_share(self): + pass -def test_is_host_valid(): - pass + def test_get_shares(self): + pass -def test_get_hosts(): - pass + def test_get_shares_by_access(self): + pass -def test_is_group_valid(): - pass + def test_get_users_with_share_access(self): + pass -def test_add_group(): - pass + def test_add_domain_backupkey(self): + pass -def test_get_groups(): - pass + def test_get_domain_backupkey(self): + pass -def test_get_group_relations(): - pass + def test_is_dpapi_secret_valid(self): + pass -def test_remove_group_relations(): - pass + def test_add_dpapi_secrets(self): + pass -def test_is_user_valid(): - pass + def test_get_dpapi_secrets(self): + pass -def test_get_users(): - pass + def test_add_loggedin_relation(self): + pass -def test_get_user(): - pass + def test_get_loggedin_relations(self): + pass -def test_get_domain_controllers(): - pass - -def test_is_share_valid(): - pass - -def test_add_share(): - pass - -def test_get_shares(): - pass - -def test_get_shares_by_access(): - pass - -def test_get_users_with_share_access(): - pass - -def test_add_domain_backupkey(): - pass - -def test_get_domain_backupkey(): - pass - -def test_is_dpapi_secret_valid(): - pass - -def test_add_dpapi_secrets(): - pass - -def test_get_dpapi_secrets(): - pass - -def test_add_loggedin_relation(): - pass - -def test_get_loggedin_relations(): - pass - -def test_remove_loggedin_relations(): - pass + def test_remove_loggedin_relations(self): + pass