Feat: create fs_datacatalogue

This commit is contained in:
2024-11-05 06:55:05 +01:00
parent 88795fdad3
commit 07fb92e2fa
10 changed files with 47 additions and 117 deletions

View File

@@ -42,12 +42,12 @@ def test_init(location):
def test_list_schema(location):
repo = FSDataCatalogue("example", location)
assert {id: s.model_dump()["id"] for id, s in repo.schemas().items()} == {
assert {id: s.model_dump()["ref"]["id"] for id, s in repo.schemas().items()} == {
".": ".",
"username": "username",
"salary": "salary",
}
assert {id: s.model_dump()["value"] for id, s in repo.schemas().items()} == {
assert {id: s.model_dump()["ref"]["value"] for id, s in repo.schemas().items()} == {
".": ".",
"username": "username",
"salary": "salary",
@@ -62,11 +62,11 @@ def test_list_schema(location):
def test_list_tables(location):
repo = FSDataCatalogue("example", location)
assert repo.tables() == {}
assert {id: t.model_dump()["value"] for id,t in repo.tables("username").items()} == {
assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("username").items()} == {
"username.csv": "username.csv",
"username-password-recovery-code.xlsx": "username-password-recovery-code.xlsx",
"username-password-recovery-code.xls": "username-password-recovery-code.xls",
}
assert {id: t.model_dump()["value"] for id,t in repo.tables("salary").items()} == {
assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("salary").items()} == {
"salary.pdf": "salary.pdf",
}

View File

@@ -1,7 +0,0 @@
Username;Identifier;First name;Last name
booker12;9012;Rachel;Booker
grey07;2070;Laura;Grey
johnson81;4081;Craig;Johnson
jenkins46;9346;Mary;Jenkins
smith79;5079;Jamie;Smith
1 Username Identifier First name Last name
2 booker12 9012 Rachel Booker
3 grey07 2070 Laura Grey
4 johnson81 4081 Craig Johnson
5 jenkins46 9346 Mary Jenkins
6 smith79 5079 Jamie Smith

View File

@@ -1,84 +0,0 @@
import shutil
from pathlib import Path
import pytest
from pandas import pandas
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",
"username-password-recovery-code.xls",
]
assert repo.tables("salary") == ["salary.pdf"]
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