314 lines
12 KiB
Python
314 lines
12 KiB
Python
import pytest
|
|
|
|
from components.repositories.db_management import RepositoriesDbManager, RepositoriesSettings, Repository, \
|
|
REPOSITORIES_SETTINGS_ENTRY
|
|
from core.settings_management import SettingsManager, MemoryDbEngine
|
|
|
|
|
|
@pytest.fixture
|
|
def settings_manager():
|
|
return SettingsManager(MemoryDbEngine())
|
|
|
|
|
|
@pytest.fixture
|
|
def db(session, settings_manager):
|
|
return RepositoriesDbManager(session, settings_manager)
|
|
|
|
|
|
@pytest.fixture
|
|
def settings_manager_with_existing_repo(db, settings_manager):
|
|
settings = RepositoriesSettings()
|
|
repo = Repository(name="ExistingRepo", tables=["Table1"])
|
|
settings.repositories.append(repo)
|
|
settings_manager.save(db.session, REPOSITORIES_SETTINGS_ENTRY, settings)
|
|
return settings_manager
|
|
|
|
|
|
def test_add_new_repository(db, settings_manager):
|
|
"""Test adding a new repository with valid data."""
|
|
db.add_repository("NewRepo", ["Table1", "Table2"])
|
|
|
|
settings = settings_manager.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert settings.repositories[0].name == "NewRepo"
|
|
assert settings.repositories[0].tables == ["Table1", "Table2"]
|
|
|
|
|
|
def test_add_repository_duplicate_name(db, settings_manager):
|
|
"""Test adding a repository with an existing name."""
|
|
settings = RepositoriesSettings()
|
|
settings.repositories.append(Repository(name="ExistingRepo", tables=[]))
|
|
settings_manager.save(db.session, REPOSITORIES_SETTINGS_ENTRY, settings)
|
|
|
|
with pytest.raises(ValueError, match="Repository 'ExistingRepo' already exists."):
|
|
db.add_repository("ExistingRepo")
|
|
|
|
|
|
def test_add_repository_empty_name(db):
|
|
"""Test adding a repository with an empty name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.add_repository("")
|
|
|
|
|
|
def test_add_repository_none_name(db):
|
|
"""Test adding a repository with a None name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.add_repository(None)
|
|
|
|
|
|
def test_add_repository_no_tables(db, settings_manager):
|
|
"""Test adding a repository without specifying tables."""
|
|
db.add_repository("RepoWithoutTables")
|
|
|
|
settings = settings_manager.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert settings.repositories[0].name == "RepoWithoutTables"
|
|
assert settings.repositories[0].tables == []
|
|
|
|
|
|
def test_get_existing_repository(db, settings_manager_with_existing_repo):
|
|
"""Test retrieving an existing repository."""
|
|
# Retrieve the repository
|
|
retrieved_repo = db.get_repository("ExistingRepo")
|
|
|
|
# Verify the repository is correctly returned
|
|
assert retrieved_repo.name == "ExistingRepo"
|
|
assert retrieved_repo.tables == ["Table1"]
|
|
|
|
|
|
def test_get_repository_not_found(db):
|
|
"""Test retrieving a repository that does not exist."""
|
|
with pytest.raises(ValueError, match="Repository 'NonExistentRepo' does not exist."):
|
|
db.get_repository("NonExistentRepo")
|
|
|
|
|
|
def test_get_repository_empty_name(db):
|
|
"""Test retrieving a repository with an empty name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.get_repository("")
|
|
|
|
|
|
def test_get_repository_none_name(db):
|
|
"""Test retrieving a repository with a None name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.get_repository(None)
|
|
|
|
|
|
def test_modify_repository_valid(db, settings_manager_with_existing_repo):
|
|
"""Test modifying an existing repository with valid data."""
|
|
modified_repo = db.modify_repository("ExistingRepo", "ModifiedRepo", ["UpdatedTable1", "UpdatedTable2"])
|
|
|
|
assert modified_repo.name == "ModifiedRepo"
|
|
assert modified_repo.tables == ["UpdatedTable1", "UpdatedTable2"]
|
|
|
|
updated_settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(updated_settings.repositories) == 1
|
|
assert updated_settings.repositories[0].name == "ModifiedRepo"
|
|
assert updated_settings.repositories[0].tables == ["UpdatedTable1", "UpdatedTable2"]
|
|
|
|
|
|
def test_modify_repository_not_found(db):
|
|
"""Test modifying a repository that does not exist."""
|
|
with pytest.raises(ValueError, match="Repository 'NonExistentRepo' not found."):
|
|
db.modify_repository("NonExistentRepo", "NewName", ["Table1"])
|
|
|
|
|
|
@pytest.mark.parametrize("repo_to_modify, new_repo", [
|
|
("", "NewName"),
|
|
(None, "NewName"),
|
|
("ExistingRepo", ""),
|
|
("ExistingRepo", None),
|
|
])
|
|
def test_modify_repository_empty_repo_to_modify(db, repo_to_modify, new_repo):
|
|
"""Test modifying a repository with an empty name for repo_to_modify."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.modify_repository(repo_to_modify, new_repo, ["Table1"])
|
|
|
|
|
|
def test_modify_repository_empty_tables_list(db, settings_manager_with_existing_repo):
|
|
"""Test modifying an existing repository to have an empty list of tables."""
|
|
modified_repo = db.modify_repository("ExistingRepo", "RepoWithEmptyTables", [])
|
|
|
|
assert modified_repo.name == "RepoWithEmptyTables"
|
|
assert modified_repo.tables == []
|
|
|
|
updated_settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(updated_settings.repositories) == 1
|
|
assert updated_settings.repositories[0].name == "RepoWithEmptyTables"
|
|
assert updated_settings.repositories[0].tables == []
|
|
|
|
|
|
def test_remove_repository_success(db, settings_manager_with_existing_repo):
|
|
"""Test successfully removing an existing repository."""
|
|
db.remove_repository("ExistingRepo")
|
|
|
|
settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 0
|
|
|
|
|
|
def test_remove_repository_not_found(db):
|
|
"""Test removing a repository that does not exist."""
|
|
with pytest.raises(ValueError, match="Repository 'NonExistentRepo' does not exist."):
|
|
db.remove_repository("NonExistentRepo")
|
|
|
|
|
|
def test_remove_repository_empty_name(db):
|
|
"""Test removing a repository with an empty name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.remove_repository("")
|
|
|
|
|
|
def test_remove_repository_none_name(db):
|
|
"""Test removing a repository with a None name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.remove_repository(None)
|
|
|
|
|
|
def test_add_table_success(db, settings_manager_with_existing_repo):
|
|
"""Test successfully adding a new table to an existing repository."""
|
|
db.add_table("ExistingRepo", "NewTable")
|
|
|
|
settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert "NewTable" in settings.repositories[0].tables
|
|
assert "Table1" in settings.repositories[0].tables
|
|
|
|
|
|
def test_add_table_repository_not_found(db):
|
|
"""Test adding a table to a non-existent repository."""
|
|
with pytest.raises(NameError, match="Repository 'NonExistentRepo' does not exist."):
|
|
db.add_table("NonExistentRepo", "NewTable")
|
|
|
|
|
|
def test_add_table_empty_name(db, settings_manager_with_existing_repo):
|
|
"""Test adding a table with an empty name."""
|
|
with pytest.raises(ValueError, match="Table name cannot be empty."):
|
|
db.add_table("ExistingRepo", "")
|
|
|
|
|
|
def test_add_table_none_name(db, settings_manager_with_existing_repo):
|
|
"""Test adding a table with a None name."""
|
|
with pytest.raises(ValueError, match="Table name cannot be empty."):
|
|
db.add_table("ExistingRepo", None)
|
|
|
|
|
|
def test_add_table_duplicate(db, settings_manager_with_existing_repo):
|
|
"""Test adding a duplicate table name."""
|
|
with pytest.raises(ValueError, match="Table 'Table1' already exists in repository 'ExistingRepo'."):
|
|
db.add_table("ExistingRepo", "Table1")
|
|
|
|
|
|
def test_modify_table_success(db, settings_manager_with_existing_repo):
|
|
"""Test successfully modifying an existing table."""
|
|
db.modify_table("ExistingRepo", "Table1", "ModifiedTable")
|
|
|
|
settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert "ModifiedTable" in settings.repositories[0].tables
|
|
assert "Table1" not in settings.repositories[0].tables
|
|
|
|
|
|
def test_modify_table_repository_not_found(db):
|
|
"""Test modifying a table in a non-existent repository."""
|
|
with pytest.raises(NameError, match="Repository 'NonExistentRepo' does not exist."):
|
|
db.modify_table("NonExistentRepo", "Table1", "NewTable")
|
|
|
|
|
|
def test_modify_table_not_found(db, settings_manager_with_existing_repo):
|
|
"""Test modifying a non-existent table."""
|
|
with pytest.raises(NameError, match="Table 'NonExistentTable' does not exist in repository 'ExistingRepo'."):
|
|
db.modify_table("ExistingRepo", "NonExistentTable", "NewTable")
|
|
|
|
|
|
@pytest.mark.parametrize("old_table, new_table", [
|
|
("", "NewTable"),
|
|
(None, "NewTable"),
|
|
("Table1", ""),
|
|
("Table1", None),
|
|
])
|
|
def test_modify_table_empty_names(db, settings_manager_with_existing_repo, old_table, new_table):
|
|
"""Test modifying a table with empty/None names."""
|
|
with pytest.raises(ValueError, match="Table name cannot be empty."):
|
|
db.modify_table("ExistingRepo", old_table, new_table)
|
|
|
|
|
|
def test_modify_table_empty_repository_name(db):
|
|
"""Test modifying a table with empty/None names."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.modify_table(None, "old_table", "new_table")
|
|
|
|
|
|
def test_remove_table_success(db, settings_manager_with_existing_repo):
|
|
"""Test successfully removing an existing table."""
|
|
db.remove_table("ExistingRepo", "Table1")
|
|
|
|
settings = settings_manager_with_existing_repo.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert "Table1" not in settings.repositories[0].tables
|
|
|
|
|
|
def test_remove_table_repository_not_found(db):
|
|
"""Test removing a table from a non-existent repository."""
|
|
with pytest.raises(NameError, match="Repository 'NonExistentRepo' does not exist."):
|
|
db.remove_table("NonExistentRepo", "Table1")
|
|
|
|
|
|
def test_remove_table_not_found(db, settings_manager_with_existing_repo):
|
|
"""Test removing a non-existent table."""
|
|
with pytest.raises(NameError, match="Table 'NonExistentTable' does not exist in repository 'ExistingRepo'."):
|
|
db.remove_table("ExistingRepo", "NonExistentTable")
|
|
|
|
|
|
def test_remove_table_empty_name(db, settings_manager_with_existing_repo):
|
|
"""Test removing a table with empty/None name."""
|
|
with pytest.raises(ValueError, match="Table name cannot be empty."):
|
|
db.remove_table("ExistingRepo", "")
|
|
with pytest.raises(ValueError, match="Table name cannot be empty."):
|
|
db.remove_table("ExistingRepo", None)
|
|
|
|
|
|
def test_remove_table_empty_repository_name(db):
|
|
"""Test removing a table with empty/None repository name."""
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.remove_table("", "Table1")
|
|
with pytest.raises(ValueError, match="Repository name cannot be empty."):
|
|
db.remove_table(None, "Table1")
|
|
|
|
def test_repository_exists(db, settings_manager):
|
|
assert db.exists_repository("SomeRepo") is False
|
|
|
|
settings = RepositoriesSettings()
|
|
repo = Repository(name="SomeRepo", tables=["Table1"])
|
|
settings.repositories.append(repo)
|
|
settings_manager.save(db.session, REPOSITORIES_SETTINGS_ENTRY, settings)
|
|
|
|
assert db.exists_repository("SomeRepo") is True
|
|
|
|
|
|
def test_repository_table(db, settings_manager_with_existing_repo):
|
|
assert db.exists_table("ExistingRepo", "SomeTable") is False
|
|
|
|
db.add_table("ExistingRepo", "SomeTable")
|
|
|
|
assert db.exists_table("ExistingRepo", "SomeTable") is True
|
|
|
|
def test_exists_table_fails_when_repo_doesnt_exist(db):
|
|
assert db.exists_table("NonExistentRepo", "SomeTable") is False
|
|
|
|
def test_ensure_exists(db, settings_manager):
|
|
settings = settings_manager.load(db.session, REPOSITORIES_SETTINGS_ENTRY, default=RepositoriesSettings())
|
|
assert len(settings.repositories) == 0
|
|
|
|
db.ensure_exists("SomeRepo", "SomeTable")
|
|
|
|
settings = settings_manager.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert settings.repositories[0].name == "SomeRepo"
|
|
assert settings.repositories[0].tables == ["SomeTable"]
|
|
|
|
db.ensure_exists("SomeRepo", "SomeTable") # as no effect when called twice
|
|
settings = settings_manager.load(db.session, REPOSITORIES_SETTINGS_ENTRY)
|
|
assert len(settings.repositories) == 1
|
|
assert settings.repositories[0].name == "SomeRepo"
|
|
assert settings.repositories[0].tables == ["SomeTable"]
|