directory, file, repository, tests
This commit is contained in:
parent
69a1aa2471
commit
383d7c57ab
GPG Key ID:
553C97999B363D38
24
pdm.lock
generated
24
pdm.lock
generated
@ -5,7 +5,29 @@
|
||||
groups = ["default", "dev"]
|
||||
strategy = ["cross_platform", "inherit_metadata"]
|
||||
lock_version = "4.4.1"
|
||||
content_hash = "sha256:1b3ad8e836b4c01729baf2a537f2c7c543495cd762f914bb7cfd518b1634ab2d"
|
||||
content_hash = "sha256:fe3214096aaef3097e2009f717762fb370bb726aa89a52e7b2a40d60016be987"
|
||||
|
||||
[[package]]
|
||||
name = "aiofiles"
|
||||
version = "24.1.0"
|
||||
requires_python = ">=3.8"
|
||||
summary = "File support for asyncio."
|
||||
groups = ["default"]
|
||||
files = [
|
||||
{file = "aiofiles-24.1.0-py3-none-any.whl", hash = "sha256:b4ec55f4195e3eb5d7abd1bf7e061763e864dd4954231fb8539a0ef8bb8260e5"},
|
||||
{file = "aiofiles-24.1.0.tar.gz", hash = "sha256:22a075c9e5a3810f0c2e48f3008c94d68c65d763b9b03857924c99e57355166c"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aioshutil"
|
||||
version = "1.5"
|
||||
requires_python = ">=3.8"
|
||||
summary = "Asynchronous shutil module."
|
||||
groups = ["default"]
|
||||
files = [
|
||||
{file = "aioshutil-1.5-py3-none-any.whl", hash = "sha256:bc2a6cdcf1a8615b62f856154fd81131031d03f2834912ebb06d8a2391253652"},
|
||||
{file = "aioshutil-1.5.tar.gz", hash = "sha256:2756d6cd3bb03405dc7348ac11a0b60eb949ebd63cdd15f56e922410231c1201"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aiosmtplib"
|
||||
|
||||
@ -34,6 +34,8 @@ dependencies = [
|
||||
"cryptography>=43.0.0",
|
||||
"python-multipart>=0.0.9",
|
||||
"jinja2>=3.1.4",
|
||||
"aiofiles>=24.1.0",
|
||||
"aioshutil>=1.5",
|
||||
]
|
||||
requires-python = ">=3.12,<3.13"
|
||||
readme = "README.md"
|
||||
|
||||
@ -2,8 +2,9 @@ from time import time
|
||||
from typing import List, Optional, Self
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import aiofiles
|
||||
|
||||
from sqlalchemy import BigInteger, ForeignKey
|
||||
from sqlalchemy import BigInteger, ForeignKey, inspect
|
||||
from sqlalchemy.orm import mapped_column, Mapped, relationship
|
||||
import sqlalchemy as sa
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
@ -11,6 +12,7 @@ from pydantic import BaseModel, ConfigDict
|
||||
from materia.models.base import Base
|
||||
from materia.models import database
|
||||
from materia.models.database import SessionContext
|
||||
from materia.config import Config
|
||||
|
||||
|
||||
class DirectoryError(Exception):
|
||||
@ -33,203 +35,159 @@ class Directory(Base):
|
||||
is_public: Mapped[bool] = mapped_column(default=False)
|
||||
|
||||
repository: Mapped["Repository"] = relationship(back_populates="directories")
|
||||
directories: Mapped[List["Directory"]] = relationship(
|
||||
back_populates="parent", remote_side=[id]
|
||||
directories: Mapped[List["Directory"]] = relationship(back_populates="parent")
|
||||
parent: Mapped["Directory"] = relationship(
|
||||
back_populates="directories", remote_side=[id]
|
||||
)
|
||||
parent: Mapped["Directory"] = relationship(back_populates="directories")
|
||||
files: Mapped[List["File"]] = relationship(back_populates="parent")
|
||||
link: Mapped["DirectoryLink"] = relationship(back_populates="directory")
|
||||
|
||||
async def new(
|
||||
self,
|
||||
session: SessionContext,
|
||||
path: Optional[Path] = None,
|
||||
with_parents: bool = False,
|
||||
) -> Optional[Self]:
|
||||
async def new(self, session: SessionContext, config: Config) -> Optional[Self]:
|
||||
session.add(self)
|
||||
await session.flush()
|
||||
await session.refresh(self, attribute_names=["repository", "parent"])
|
||||
await session.refresh(self, attribute_names=["repository"])
|
||||
|
||||
repository_path: Path = await self.repository.path(session)
|
||||
current_path: Path = repository_path
|
||||
current_directory: Optional[Directory] = None
|
||||
|
||||
for part in path.parts:
|
||||
current_path /= part
|
||||
relative_path = current_path.relative_to(repository_path)
|
||||
|
||||
if current_path.exists() and current_path.is_dir():
|
||||
# Find record
|
||||
current_directory = await Directory.find(
|
||||
self.repository, self.parent, self.name, session
|
||||
)
|
||||
|
||||
if not current_directory:
|
||||
# TODO: recreate record
|
||||
raise DirectoryError(
|
||||
f"No directory was found in the records: {relative_path}"
|
||||
)
|
||||
|
||||
current_directory.updated = time()
|
||||
await session.flush()
|
||||
|
||||
continue
|
||||
|
||||
if not with_parents:
|
||||
raise DirectoryError(f"Directory not exists at /{relative_path}")
|
||||
|
||||
# Create an ancestor directory from scratch
|
||||
current_directory = await Directory(
|
||||
repository_id=self.repository.id,
|
||||
parent_id=current_directory.id if current_directory else None,
|
||||
name=part,
|
||||
).new(
|
||||
session,
|
||||
path=relative_path,
|
||||
with_parents=False,
|
||||
)
|
||||
|
||||
try:
|
||||
current_path.mkdir()
|
||||
except OSError as e:
|
||||
raise DirectoryError(
|
||||
f"Failed to create directory at /{relative_path}:", *e.args
|
||||
)
|
||||
|
||||
# Create directory
|
||||
current_path /= self.name
|
||||
relative_path = current_path.relative_to(repository_path)
|
||||
relative_path = await self.relative_path(session)
|
||||
directory_path = await self.path(session, config)
|
||||
|
||||
try:
|
||||
current_path.mkdir()
|
||||
directory_path.mkdir()
|
||||
except OSError as e:
|
||||
raise DirectoryError(
|
||||
f"Failed to create directory at /{relative_path}:", *e.args
|
||||
f"Failed to create directory at /{relative_path}:",
|
||||
*e.args,
|
||||
)
|
||||
|
||||
# Update information
|
||||
self.parent = current_directory
|
||||
|
||||
await session.commit()
|
||||
|
||||
return self
|
||||
|
||||
async def remove(self, session: SessionContext):
|
||||
async def remove(self, session: SessionContext, config: Config):
|
||||
session.add(self)
|
||||
await session.refresh(self, attribute_names=["directories", "files"])
|
||||
|
||||
current_path: Path = self.repository.path(session) / self.path(session)
|
||||
if self.directories:
|
||||
for directory in self.directories:
|
||||
directory.remove(session, config)
|
||||
|
||||
if self.files:
|
||||
for file in self.files:
|
||||
file.remove(session, config)
|
||||
|
||||
relative_path = await self.relative_path(session)
|
||||
directory_path = await self.path(session, config)
|
||||
|
||||
try:
|
||||
shutil.tmtree(str(current_path))
|
||||
shutil.rmtree(str(directory_path))
|
||||
except OSError as e:
|
||||
raise DirectoryError("Failed to remove directory:", *e.args)
|
||||
|
||||
await session.refresh(self, attribute_names=["parent"])
|
||||
current_directory: Directory = self.parent
|
||||
|
||||
while current_directory:
|
||||
current_directory.updated = time()
|
||||
session.add(current_directory)
|
||||
await session.refresh(self, attribute_names=["parent"])
|
||||
current_directory = current_directory.parent
|
||||
raise DirectoryError(
|
||||
f"Failed to remove directory at /{relative_path}:", *e.args
|
||||
)
|
||||
|
||||
await session.delete(self)
|
||||
await session.commit()
|
||||
await session.flush()
|
||||
|
||||
async def is_root(self) -> bool:
|
||||
return self.parent_id is None
|
||||
|
||||
@staticmethod
|
||||
async def find(
|
||||
repository: "Repository",
|
||||
directory: "Directory",
|
||||
name: str,
|
||||
session: SessionContext,
|
||||
) -> Optional[Self]:
|
||||
return (
|
||||
await session.scalars(
|
||||
sa.select(Directory).where(
|
||||
sa.and_(
|
||||
Directory.repository_id == repository.id,
|
||||
Directory.name == name,
|
||||
Directory.parent_id == directory.parent_id,
|
||||
)
|
||||
)
|
||||
)
|
||||
).first()
|
||||
|
||||
async def find_nested(self, session: SessionContext):
|
||||
pass
|
||||
|
||||
async def find_by_descend(
|
||||
self, path: Path | str, db: database.Database, need_create: bool = False
|
||||
) -> Optional[Self]:
|
||||
"""Find a nested directory from current"""
|
||||
repository_id = self.repository_id
|
||||
path = Path(path)
|
||||
current_directory = self
|
||||
|
||||
async with db.session() as session:
|
||||
for part in path.parts:
|
||||
directory = (
|
||||
await session.scalars(
|
||||
sa.select(Directory).where(
|
||||
sa.and_(
|
||||
Directory.repository_id == repository_id,
|
||||
Directory.name == part,
|
||||
Directory.parent_id == current_directory.id,
|
||||
)
|
||||
)
|
||||
)
|
||||
).first()
|
||||
|
||||
if directory is None:
|
||||
if not need_create:
|
||||
return None
|
||||
|
||||
directory = Directory(
|
||||
repository_id=repository_id,
|
||||
parent_id=current_directory.id,
|
||||
name=part,
|
||||
)
|
||||
session.add(directory)
|
||||
await session.flush()
|
||||
|
||||
current_directory = directory
|
||||
|
||||
if need_create:
|
||||
await session.commit()
|
||||
|
||||
return current_directory
|
||||
|
||||
@staticmethod
|
||||
async def find_by_path(
|
||||
repository_id: int, path: Path | str, db: database.Database
|
||||
) -> Optional[Self]:
|
||||
"""Find a directory by given absolute path"""
|
||||
path = Path(path)
|
||||
assert path == Path(), "The path cannot be empty"
|
||||
|
||||
root = await Directory.find_by_descend(repository_id, path.parts[0], db)
|
||||
return root.descend(Path().joinpath(*path.parts[1:]), db)
|
||||
|
||||
async def path(self, session: SessionContext) -> Optional[Path]:
|
||||
async def relative_path(self, session: SessionContext) -> Optional[Path]:
|
||||
"""Get relative path of the current directory"""
|
||||
if inspect(self).was_deleted:
|
||||
return None
|
||||
|
||||
parts = []
|
||||
current_directory = self
|
||||
|
||||
while True:
|
||||
parts.append(current_directory.name)
|
||||
session.add(current_directory)
|
||||
await session.refresh(current_directory, attribute_names=["parent"])
|
||||
async with session.begin_nested():
|
||||
while True:
|
||||
parts.append(current_directory.name)
|
||||
|
||||
if current_directory.parent is None:
|
||||
break
|
||||
session.add(current_directory)
|
||||
await session.refresh(current_directory, attribute_names=["parent"])
|
||||
|
||||
current_directory = current_directory.parent
|
||||
if current_directory.parent is None:
|
||||
break
|
||||
|
||||
current_directory = current_directory.parent
|
||||
|
||||
return Path().joinpath(*reversed(parts))
|
||||
|
||||
async def path(self, session: SessionContext, config: Config) -> Optional[Path]:
|
||||
if inspect(self).was_deleted:
|
||||
return None
|
||||
|
||||
repository_path = await self.repository.path(session, config)
|
||||
relative_path = await self.relative_path(session)
|
||||
|
||||
return repository_path.joinpath(relative_path)
|
||||
|
||||
def is_root(self) -> bool:
|
||||
return self.parent_id is None
|
||||
|
||||
@staticmethod
|
||||
async def by_path(
|
||||
repository: "Repository", path: Path, session: SessionContext, config: Config
|
||||
) -> Optional[Self]:
|
||||
if path == Path():
|
||||
raise DirectoryError("Cannot find directory by empty path")
|
||||
|
||||
current_directory: Optional[Directory] = None
|
||||
|
||||
for part in path.parts:
|
||||
current_directory = (
|
||||
await session.scalars(
|
||||
sa.select(Directory).where(
|
||||
sa.and_(
|
||||
Directory.repository_id == repository.id,
|
||||
Directory.name == part,
|
||||
(
|
||||
Directory.parent_id == current_directory.id
|
||||
if current_directory
|
||||
else Directory.parent_id.is_(None)
|
||||
),
|
||||
)
|
||||
)
|
||||
)
|
||||
).first()
|
||||
|
||||
if not current_directory:
|
||||
return None
|
||||
|
||||
return current_directory
|
||||
|
||||
async def copy(
|
||||
self, directory: Optional["Directory"], session: SessionContext, config: Config
|
||||
) -> Self:
|
||||
pass
|
||||
|
||||
async def move(
|
||||
self, directory: Optional["Directory"], session: SessionContext, config: Config
|
||||
) -> Self:
|
||||
pass
|
||||
|
||||
async def rename(self, name: str, session: SessionContext, config: Config) -> Self:
|
||||
session.add(self)
|
||||
|
||||
directory_path = await self.path(session, config)
|
||||
relative_path = await self.relative_path(session)
|
||||
new_path = directory_path.with_name(name)
|
||||
identity = 1
|
||||
|
||||
while True:
|
||||
if new_path == directory_path:
|
||||
break
|
||||
if not new_path.exists():
|
||||
break
|
||||
|
||||
new_path = directory_path.with_name(f"{name}.{str(identity)}")
|
||||
identity += 1
|
||||
|
||||
try:
|
||||
await aiofiles.os.rename(directory_path, new_path)
|
||||
except OSError as e:
|
||||
raise DirectoryError(
|
||||
f"Failed to rename directory at /{relative_path}", *e.args
|
||||
)
|
||||
|
||||
self.name = new_path.name
|
||||
await session.flush()
|
||||
return self
|
||||
|
||||
async def info(self) -> "DirectoryInfo":
|
||||
return DirectoryInfo.model_validate(self)
|
||||
|
||||
|
||||
@ -1,14 +1,23 @@
|
||||
from time import time
|
||||
from typing import Optional, Self
|
||||
from pathlib import Path
|
||||
import aioshutil
|
||||
|
||||
from sqlalchemy import BigInteger, ForeignKey
|
||||
from sqlalchemy import BigInteger, ForeignKey, inspect
|
||||
from sqlalchemy.orm import mapped_column, Mapped, relationship
|
||||
import sqlalchemy as sa
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
import aiofiles
|
||||
import aiofiles.os
|
||||
|
||||
from materia.models.base import Base
|
||||
from materia.models import database
|
||||
from materia.models.database import SessionContext
|
||||
from materia.config import Config
|
||||
|
||||
|
||||
class FileError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class File(Base):
|
||||
@ -24,40 +33,184 @@ class File(Base):
|
||||
created: Mapped[int] = mapped_column(BigInteger, nullable=False, default=time)
|
||||
updated: Mapped[int] = mapped_column(BigInteger, nullable=False, default=time)
|
||||
name: Mapped[str]
|
||||
path: Mapped[str] = mapped_column(nullable=True)
|
||||
is_public: Mapped[bool] = mapped_column(default=False)
|
||||
size: Mapped[int] = mapped_column(BigInteger)
|
||||
size: Mapped[int] = mapped_column(BigInteger, nullable=True)
|
||||
|
||||
repository: Mapped["Repository"] = relationship(back_populates="files")
|
||||
parent: Mapped["Directory"] = relationship(back_populates="files")
|
||||
link: Mapped["FileLink"] = relationship(back_populates="file")
|
||||
|
||||
async def new(
|
||||
self, data: bytes, session: SessionContext, config: Config
|
||||
) -> Optional[Self]:
|
||||
session.add(self)
|
||||
await session.flush()
|
||||
await session.refresh(self, attribute_names=["repository"])
|
||||
|
||||
relative_path = await self.relative_path(session)
|
||||
file_path = await self.path(session, config)
|
||||
size = None
|
||||
|
||||
try:
|
||||
async with aiofiles.open(file_path, mode="wb") as file:
|
||||
await file.write(data)
|
||||
size = (await aiofiles.os.stat(file_path)).st_size
|
||||
except OSError as e:
|
||||
raise FileError(f"Failed to write file at /{relative_path}", *e.args)
|
||||
|
||||
self.size = size
|
||||
await session.flush()
|
||||
|
||||
return self
|
||||
|
||||
async def remove(self, session: SessionContext, config: Config):
|
||||
session.add(self)
|
||||
|
||||
relative_path = await self.relative_path(session)
|
||||
file_path = await self.path(session, config)
|
||||
|
||||
try:
|
||||
await aiofiles.os.remove(file_path)
|
||||
except OSError as e:
|
||||
raise FileError(f"Failed to remove file at /{relative_path}:", *e.args)
|
||||
|
||||
await session.delete(self)
|
||||
await session.flush()
|
||||
|
||||
async def relative_path(self, session: SessionContext) -> Optional[Path]:
|
||||
if inspect(self).was_deleted:
|
||||
return None
|
||||
|
||||
file_path = Path()
|
||||
|
||||
async with session.begin_nested():
|
||||
session.add(self)
|
||||
await session.refresh(self, attribute_names=["parent"])
|
||||
|
||||
if self.parent:
|
||||
file_path = await self.parent.relative_path(session)
|
||||
|
||||
return file_path.joinpath(self.name)
|
||||
|
||||
async def path(self, session: SessionContext, config: Config) -> Optional[Path]:
|
||||
if inspect(self).was_deleted:
|
||||
return None
|
||||
|
||||
file_path = Path()
|
||||
|
||||
async with session.begin_nested():
|
||||
session.add(self)
|
||||
await session.refresh(self, attribute_names=["repository", "parent"])
|
||||
|
||||
if self.parent:
|
||||
file_path = await self.parent.path(session, config)
|
||||
else:
|
||||
file_path = await self.repository.path(session, config)
|
||||
|
||||
return file_path.joinpath(self.name)
|
||||
|
||||
@staticmethod
|
||||
async def by_path(
|
||||
repository_id: int, path: Path | None, name: str, db: database.Database
|
||||
) -> Self | None:
|
||||
async with db.session() as session:
|
||||
query_path = (
|
||||
File.path == str(path)
|
||||
if isinstance(path, Path)
|
||||
else File.path.is_(None)
|
||||
)
|
||||
return (
|
||||
await session.scalars(
|
||||
sa.select(File).where(
|
||||
sa.and_(
|
||||
File.repository_id == repository_id,
|
||||
File.name == name,
|
||||
query_path,
|
||||
)
|
||||
repository: "Repository", path: Path, session: SessionContext, config: Config
|
||||
) -> Optional[Self]:
|
||||
if path == Path():
|
||||
raise FileError("Cannot find file by empty path")
|
||||
|
||||
parent_directory = await Directory.by_path(
|
||||
repository, path.parent, session, config
|
||||
)
|
||||
|
||||
current_file = (
|
||||
await session.scalars(
|
||||
sa.select(File).where(
|
||||
sa.and_(
|
||||
File.repository_id == repository.id,
|
||||
File.name == path.name,
|
||||
(
|
||||
File.parent_id == parent_directory.id
|
||||
if parent_directory
|
||||
else File.parent_id.is_(None)
|
||||
),
|
||||
)
|
||||
)
|
||||
).first()
|
||||
)
|
||||
).first()
|
||||
|
||||
async def remove(self, db: database.Database):
|
||||
async with db.session() as session:
|
||||
await session.delete(self)
|
||||
await session.commit()
|
||||
return current_file
|
||||
|
||||
async def copy(
|
||||
self, directory: Optional["Directory"], session: SessionContext, config: Config
|
||||
) -> Self:
|
||||
pass
|
||||
|
||||
async def move(
|
||||
self, directory: Optional["Directory"], session: SessionContext, config: Config
|
||||
) -> Self:
|
||||
session.add(self)
|
||||
await session.refresh(self, attribute_names=["repository"])
|
||||
|
||||
repository_path = await self.repository.path(session, config)
|
||||
file_path = await self.path(session, config)
|
||||
directory_path = (
|
||||
await directory.path(session, config) if directory else repository_path
|
||||
)
|
||||
new_path = File.generate_name(file_path, directory_path, self.name)
|
||||
|
||||
try:
|
||||
await aioshutil.move(file_path, new_path)
|
||||
except OSError as e:
|
||||
raise FileError("Failed to move file:", *e.args)
|
||||
|
||||
self.parent_id = directory.id if directory else None
|
||||
await session.flush()
|
||||
|
||||
return self
|
||||
|
||||
@staticmethod
|
||||
def generate_name(old_file: Path, target_directory: Path, name: str) -> Path:
|
||||
new_path = target_directory.joinpath(name)
|
||||
identity = 1
|
||||
|
||||
while True:
|
||||
if new_path == old_file:
|
||||
break
|
||||
if not new_path.exists():
|
||||
break
|
||||
|
||||
new_path = target_directory.joinpath(
|
||||
f"{name.removesuffix(new_path.suffix)}.{str(identity)}{new_path.suffix}"
|
||||
)
|
||||
identity += 1
|
||||
|
||||
return new_path
|
||||
|
||||
async def rename(self, name: str, session: SessionContext, config: Config) -> Self:
|
||||
session.add(self)
|
||||
|
||||
file_path = await self.path(session, config)
|
||||
relative_path = await self.relative_path(session)
|
||||
new_path = File.generate_name(file_path, file_path.parent, name)
|
||||
|
||||
try:
|
||||
await aiofiles.os.rename(file_path, new_path)
|
||||
except OSError as e:
|
||||
raise FileError(f"Failed to rename file at /{relative_path}", *e.args)
|
||||
|
||||
self.name = new_path.name
|
||||
await session.flush()
|
||||
return self
|
||||
|
||||
async def info(self) -> Optional["FileInfo"]:
|
||||
if self.is_public:
|
||||
return FileInfo.model_validate(self)
|
||||
return None
|
||||
|
||||
|
||||
def convert_bytes(size: int):
|
||||
for unit in ["bytes", "kB", "MB", "GB", "TB"]:
|
||||
if size < 1024:
|
||||
return f"{size}{unit}" if unit == "bytes" else f"{size:.1f}{unit}"
|
||||
size >>= 10
|
||||
|
||||
|
||||
class FileLink(Base):
|
||||
@ -80,7 +233,6 @@ class FileInfo(BaseModel):
|
||||
created: int
|
||||
updated: int
|
||||
name: str
|
||||
path: Optional[str]
|
||||
is_public: bool
|
||||
size: int
|
||||
|
||||
|
||||
@ -34,13 +34,17 @@ class Repository(Base):
|
||||
async def new(self, session: SessionContext, config: Config) -> Optional[Self]:
|
||||
session.add(self)
|
||||
await session.flush()
|
||||
|
||||
repository_path = await self.path(session, config)
|
||||
relative_path = repository_path.relative_to(
|
||||
config.application.working_directory
|
||||
)
|
||||
|
||||
try:
|
||||
repository_path.mkdir(parents=True, exist_ok=True)
|
||||
except OSError as e:
|
||||
raise RepositoryError(
|
||||
f"Failed to create repository at /{repository_path.relative_to(config.application.working_directory)}:",
|
||||
f"Failed to create repository at /{relative_path}:",
|
||||
*e.args,
|
||||
)
|
||||
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import pytest_asyncio
|
||||
import pytest
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from materia.config import Config
|
||||
from materia.models import (
|
||||
@ -10,6 +11,7 @@ from materia.models import (
|
||||
Repository,
|
||||
Directory,
|
||||
RepositoryError,
|
||||
File,
|
||||
)
|
||||
from materia.models.base import Base
|
||||
from materia.models.database import SessionContext
|
||||
@ -17,7 +19,9 @@ from materia import security
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.pool import NullPool
|
||||
from sqlalchemy.orm.session import make_transient
|
||||
from dataclasses import dataclass
|
||||
from sqlalchemy import inspect
|
||||
import aiofiles
|
||||
import aiofiles.os
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session")
|
||||
@ -188,17 +192,19 @@ async def test_repository(data, tmpdir, session: SessionContext, config: Config)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_directory(data, tmpdir, session: SessionContext, config: Config):
|
||||
config.application.working_directory = Path(tmpdir)
|
||||
|
||||
# setup
|
||||
session.add(data.user)
|
||||
await session.flush()
|
||||
|
||||
repository = Repository(user_id=data.user.id, capacity=config.repository.capacity)
|
||||
session.add(repository)
|
||||
await session.flush()
|
||||
repository = await Repository(
|
||||
user_id=data.user.id, capacity=config.repository.capacity
|
||||
).new(session, config)
|
||||
|
||||
directory = Directory(repository_id=repository.id, parent_id=None, name="test1")
|
||||
session.add(directory)
|
||||
await session.flush()
|
||||
directory = await Directory(
|
||||
repository_id=repository.id, parent_id=None, name="test1"
|
||||
).new(session, config)
|
||||
|
||||
# simple
|
||||
assert directory.id is not None
|
||||
@ -212,15 +218,14 @@ async def test_directory(data, tmpdir, session: SessionContext, config: Config):
|
||||
)
|
||||
)
|
||||
).first() == directory
|
||||
assert (await directory.path(session, config)).exists()
|
||||
|
||||
# nested simple
|
||||
nested_directory = Directory(
|
||||
nested_directory = await Directory(
|
||||
repository_id=repository.id,
|
||||
parent_id=directory.id,
|
||||
name="test_nested",
|
||||
)
|
||||
session.add(nested_directory)
|
||||
await session.flush()
|
||||
).new(session, config)
|
||||
|
||||
assert nested_directory.id is not None
|
||||
assert (
|
||||
@ -234,3 +239,153 @@ async def test_directory(data, tmpdir, session: SessionContext, config: Config):
|
||||
)
|
||||
).first() == nested_directory
|
||||
assert nested_directory.parent_id == directory.id
|
||||
assert (await nested_directory.path(session, config)).exists()
|
||||
|
||||
# relationship
|
||||
await session.refresh(directory, attribute_names=["directories", "files"])
|
||||
assert isinstance(directory.files, list) and len(directory.files) == 0
|
||||
assert isinstance(directory.directories, list) and len(directory.directories) == 1
|
||||
|
||||
await session.refresh(nested_directory, attribute_names=["directories", "files"])
|
||||
assert (nested_directory.files, list) and len(nested_directory.files) == 0
|
||||
assert (nested_directory.directories, list) and len(
|
||||
nested_directory.directories
|
||||
) == 0
|
||||
|
||||
#
|
||||
assert (
|
||||
await Directory.by_path(
|
||||
repository, Path("test1", "test_nested"), session, config
|
||||
)
|
||||
== nested_directory
|
||||
)
|
||||
|
||||
# remove nested
|
||||
nested_path = await nested_directory.path(session, config)
|
||||
assert nested_path.exists()
|
||||
await nested_directory.remove(session, config)
|
||||
assert inspect(nested_directory).was_deleted
|
||||
assert await nested_directory.path(session, config) is None
|
||||
assert not nested_path.exists()
|
||||
|
||||
await session.refresh(directory) # update attributes that was deleted
|
||||
assert (await directory.path(session, config)).exists()
|
||||
|
||||
# rename
|
||||
assert (await directory.rename("test1", session, config)).name == "test1"
|
||||
directory2 = await Directory(
|
||||
repository_id=repository.id, parent_id=None, name="test2"
|
||||
).new(session, config)
|
||||
assert (await directory.rename("test2", session, config)).name == "test2.1"
|
||||
assert (await repository.path(session, config)).joinpath("test2.1").exists()
|
||||
assert not (await repository.path(session, config)).joinpath("test1").exists()
|
||||
|
||||
directory_path = await directory.path(session, config)
|
||||
assert directory_path.exists()
|
||||
|
||||
await directory.remove(session, config)
|
||||
assert await directory.path(session, config) is None
|
||||
assert not directory_path.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file(data, tmpdir, session: SessionContext, config: Config):
|
||||
config.application.working_directory = Path(tmpdir)
|
||||
|
||||
# setup
|
||||
session.add(data.user)
|
||||
await session.flush()
|
||||
|
||||
repository = await Repository(
|
||||
user_id=data.user.id, capacity=config.repository.capacity
|
||||
).new(session, config)
|
||||
|
||||
directory = await Directory(
|
||||
repository_id=repository.id, parent_id=None, name="test1"
|
||||
).new(session, config)
|
||||
directory2 = await Directory(
|
||||
repository_id=repository.id, parent_id=None, name="test2"
|
||||
).new(session, config)
|
||||
|
||||
data = b"Hello there, it's a test"
|
||||
file = await File(
|
||||
repository_id=repository.id,
|
||||
parent_id=directory.id,
|
||||
name="test_file.txt",
|
||||
).new(data, session, config)
|
||||
|
||||
# simple
|
||||
assert file.id is not None
|
||||
assert (
|
||||
await session.scalars(
|
||||
sa.select(File).where(
|
||||
sa.and_(
|
||||
File.repository_id == repository.id,
|
||||
File.parent_id == directory.id,
|
||||
File.name == "test_file.txt",
|
||||
)
|
||||
)
|
||||
)
|
||||
).first() == file
|
||||
|
||||
# relationship
|
||||
await session.refresh(file, attribute_names=["parent", "repository"])
|
||||
assert file.parent == directory
|
||||
assert file.repository == repository
|
||||
|
||||
#
|
||||
assert (
|
||||
await File.by_path(repository, Path("test1", "test_file.txt"), session, config)
|
||||
== file
|
||||
)
|
||||
|
||||
#
|
||||
file_path = await file.path(session, config)
|
||||
assert file_path.exists()
|
||||
assert (await aiofiles.os.stat(file_path)).st_size == file.size
|
||||
async with aiofiles.open(file_path, mode="rb") as io:
|
||||
content = await io.read()
|
||||
assert data == content
|
||||
|
||||
# rename
|
||||
assert (
|
||||
await file.rename("test_file_rename.txt", session, config)
|
||||
).name == "test_file_rename.txt"
|
||||
file2 = await File(
|
||||
repository_id=repository.id, parent_id=directory.id, name="test_file_2.txt"
|
||||
).new(b"", session, config)
|
||||
assert (
|
||||
await file.rename("test_file_2.txt", session, config)
|
||||
).name == "test_file_2.1.txt"
|
||||
assert (
|
||||
(await repository.path(session, config))
|
||||
.joinpath("test1", "test_file_2.1.txt")
|
||||
.exists()
|
||||
)
|
||||
assert (
|
||||
not (await repository.path(session, config))
|
||||
.joinpath("test1", "test_file_rename.txt")
|
||||
.exists()
|
||||
)
|
||||
|
||||
# move
|
||||
await file.move(directory2, session, config)
|
||||
await session.refresh(file, attribute_names=["parent"])
|
||||
assert file.parent == directory2
|
||||
assert (
|
||||
not (await repository.path(session, config))
|
||||
.joinpath("test1", "test_file_2.1.txt")
|
||||
.exists()
|
||||
)
|
||||
assert (
|
||||
(await repository.path(session, config))
|
||||
.joinpath("test2", "test_file_2.1.txt")
|
||||
.exists()
|
||||
)
|
||||
|
||||
# remove
|
||||
await file.remove(session, config)
|
||||
assert not await File.by_path(
|
||||
repository, Path("test1", "test_file.txt"), session, config
|
||||
)
|
||||
assert not file_path.exists()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user