swisskyrepo
/
NetExec
mirror of https://github.com/swisskyrepo/NetExec.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tests(smb): add setup, teardown, initial add_host test, and definitions for all smb function tests

main
Marshall Hallenbeck 2023-03-20 21:06:21 -04:00 committed by Marshall Hallenbeck
parent b7e92281ab
commit ae4d64746a
1 changed files with 102 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

260
tests/test_smb_database.py
View File

@ -4,205 +4,149 @@ import asyncio
import os import os
from time import sleep from time import sleep
import pytest from cme.cmedb import CMEDBMenu, create_workspace, create_db_engine, delete_workspace
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.first_run import first_run_setup from cme.first_run import first_run_setup
from cme.loaders.protocol_loader import protocol_loader from cme.loaders.protocol_loader import protocol_loader
from cme.logger import setup_logger, CMEAdapter from cme.logger import setup_logger, CMEAdapter
from cme.paths import WS_PATH from cme.paths import CONFIG_PATH, WS_PATH
from sqlalchemy.dialects.sqlite import Insert
@pytest.fixture(scope="session") class TestSmbDatabase:
def event_loop(): def setup_class(self):
return asyncio.get_event_loop() 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 teardown_class(self):
def db_engine(): asyncio.run(self.db.shutdown_db())
db_path = os.path.join(WS_PATH, "test/smb.db") delete_workspace("test")
db_engine = create_engine(
f"sqlite:///{db_path}",
isolation_level="AUTOCOMMIT",
future=True
)
yield db_engine
db_engine.dispose()
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") def test_update_host(self):
async def db(db_engine): pass
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"] def test_add_credential(self):
protocol_db_object = getattr(p_loader.load_protocol(protocol_db_path), "database") pass
db = protocol_db_object(db_engine) def test_update_credential(self):
yield db pass
db.shutdown_db()
delete_workspace("test")
def test_remove_credential(self):
pass
@pytest_asyncio.fixture(scope="session") def test_add_admin_user(self):
def sess(db_engine): pass
session_factory = sessionmaker(
bind=db_engine,
expire_on_commit=True
)
Session = scoped_session( def test_get_admin_relations(self):
session_factory pass
)
sess = Session()
yield sess
sess.close()
def test_remove_admin_relation(self):
pass
@pytest.mark.asyncio def test_is_credential_valid(self):
async def test_add_host(db): pass
await db.add_host(
"127.0.0.1",
"localhost",
"TEST.DEV",
"Windows Testing 2023",
False,
True,
True,
True,
False,
False
)
def test_get_credentials(self):
pass
def test_update_host(db, sess): def test_get_credential(self):
host = { pass
"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_is_credential_local(self):
pass
def test_add_credential(): def test_is_host_valid(self):
pass pass
def test_get_hosts(self):
pass
def test_update_credential(): def test_is_group_valid(self):
pass pass
def test_add_group(self):
pass
def test_remove_credential(): def test_get_groups(self):
pass pass
def test_get_group_relations(self):
pass
def test_add_admin_user(): def test_remove_group_relations(self):
pass pass
def test_get_admin_relations(): def test_is_user_valid(self):
pass pass
def test_remove_admin_relation(): def test_get_users(self):
pass pass
def test_is_credential_valid(): def test_get_user(self):
pass pass
def test_get_credentials(): def test_get_domain_controllers(self):
pass pass
def test_get_credential(): def test_is_share_valid(self):
pass pass
def test_is_credential_local(): def test_add_share(self):
pass pass
def test_is_host_valid(): def test_get_shares(self):
pass pass
def test_get_hosts(): def test_get_shares_by_access(self):
pass pass
def test_is_group_valid(): def test_get_users_with_share_access(self):
pass pass
def test_add_group(): def test_add_domain_backupkey(self):
pass pass
def test_get_groups(): def test_get_domain_backupkey(self):
pass pass
def test_get_group_relations(): def test_is_dpapi_secret_valid(self):
pass pass
def test_remove_group_relations(): def test_add_dpapi_secrets(self):
pass pass
def test_is_user_valid(): def test_get_dpapi_secrets(self):
pass pass
def test_get_users(): def test_add_loggedin_relation(self):
pass pass
def test_get_user(): def test_get_loggedin_relations(self):
pass pass
def test_get_domain_controllers(): def test_remove_loggedin_relations(self):
pass 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