2024-08-14 08:44:38 +00:00
|
|
|
import shutil
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
import pytest
|
2024-10-07 03:27:46 +00:00
|
|
|
from pandas import pandas
|
2024-08-14 08:44:38 +00:00
|
|
|
|
|
|
|
from dashboard.libs.repository.fs_repository import FSRepository
|
|
|
|
|
|
|
|
EXAMPLE_DIR = "./tests/repository/fs_examples/"
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def location(tmp_path):
|
|
|
|
loc = tmp_path
|
|
|
|
username_loc = loc / "username"
|
|
|
|
username_loc.mkdir()
|
|
|
|
salary_loc = loc / "salary"
|
|
|
|
salary_loc.mkdir()
|
|
|
|
example_src = Path(EXAMPLE_DIR)
|
|
|
|
|
|
|
|
for f in example_src.glob("*"):
|
|
|
|
if "username" in str(f):
|
|
|
|
shutil.copy(f, username_loc)
|
|
|
|
else:
|
|
|
|
shutil.copy(f, salary_loc)
|
|
|
|
|
|
|
|
return loc
|
|
|
|
|
|
|
|
|
|
|
|
def test_init(location):
|
|
|
|
repo = FSRepository("example", location)
|
|
|
|
assert repo.ls() == [
|
|
|
|
"username",
|
|
|
|
"salary",
|
|
|
|
]
|
|
|
|
assert repo.schemas() == [
|
|
|
|
".",
|
|
|
|
"username",
|
|
|
|
"salary",
|
|
|
|
]
|
|
|
|
|
|
|
|
assert repo.tables() == []
|
|
|
|
assert repo.tables("username") == [
|
|
|
|
"username.csv",
|
|
|
|
"username-password-recovery-code.xlsx",
|
2024-10-07 03:27:46 +00:00
|
|
|
"username-password-recovery-code.xls",
|
2024-08-14 08:44:38 +00:00
|
|
|
]
|
|
|
|
assert repo.tables("salary") == ["salary.pdf"]
|
2024-10-07 03:27:46 +00:00
|
|
|
|
|
|
|
|
|
|
|
def test_read_csv(location):
|
|
|
|
repo = FSRepository("example", location)
|
|
|
|
username = repo.read("username.csv", "username", delimiter=";")
|
|
|
|
assert list(username.columns) == [
|
|
|
|
"Username",
|
|
|
|
"Identifier",
|
|
|
|
"First name",
|
|
|
|
"Last name",
|
|
|
|
]
|
|
|
|
assert len(username.index) == 5
|
|
|
|
|
|
|
|
|
|
|
|
def test_fake_read_xlsx(location):
|
|
|
|
repo = FSRepository("example", location)
|
|
|
|
df = pandas.read_excel(
|
|
|
|
location / "username" / "username-password-recovery-code.xls"
|
|
|
|
)
|
|
|
|
print(df)
|
|
|
|
|
|
|
|
|
|
|
|
def test_read_xlsx(location):
|
|
|
|
repo = FSRepository("example", location)
|
|
|
|
username = repo.read("username-password-recovery-code.xls", "username")
|
|
|
|
assert list(username.columns) == [
|
|
|
|
"Username",
|
|
|
|
"Identifier",
|
|
|
|
"One-time password",
|
|
|
|
"Recovery code",
|
|
|
|
"First name",
|
|
|
|
"Last name",
|
|
|
|
"Department",
|
|
|
|
"Location",
|
|
|
|
]
|
|
|
|
assert len(username.index) == 5
|