MyFastHtml/tests/core/test_db_object.py

import shutil
from dataclasses import dataclass

import pytest

from myfasthtml.core.dbmanager import DbManager, DbObject


@pytest.fixture(scope="session")
def session():
  return {
      "user_info": {
          "id": "test_tenant_id",
          "email": "test@email.com",
          "username": "test user",
          "role": [],
      }
  }


@pytest.fixture
def db_manager(session):
  shutil.rmtree("TestDb", ignore_errors=True)
  db_manager_instance = DbManager(session, root="TestDb", auto_register=False)
  
  yield db_manager_instance
  
  shutil.rmtree("TestDb", ignore_errors=True)


def simplify(res: dict) -> dict:
  return {k: v for k, v in res.items() if not k.startswith("_")}


def test_i_can_init(session, db_manager):
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
      
      with self.initializing():
        self.value: str = "hello"
        self.number: int = 42
        self.none_value: None = None
  
  dummy = DummyObject(session)
  
  props = dummy._get_properties()
  
  in_db = db_manager.load("DummyObject")
  history = db_manager.db.history(db_manager.get_tenant(), "DummyObject")
  assert simplify(in_db) == {"value": "hello", "number": 42, "none_value": None}
  assert len(history) == 1


def test_i_can_init_from_dataclass(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
    none_value: None = None
  
  DummyObject(session)
  
  in_db = db_manager.load("DummyObject")
  history = db_manager.db.history(db_manager.get_tenant(), "DummyObject")
  assert simplify(in_db) == {"value": "hello", "number": 42, "none_value": None}
  assert len(history) == 1


def test_i_can_init_from_db(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
  
  # insert other values in db
  db_manager.save("DummyObject", {"value": "other_value", "number": 34})
  
  dummy = DummyObject(session)
  
  assert dummy.value == "other_value"
  assert dummy.number == 34


def test_db_is_updated_when_attribute_is_modified(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
  
  dummy = DummyObject(session)
  dummy.value = "other_value"
  
  assert simplify(db_manager.load("DummyObject")) == {"value": "other_value", "number": 42}


def test_i_do_not_save_in_db_when_value_is_the_same(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
  
  dummy = DummyObject(session)
  dummy.value = "other_value"
  in_db_1 = db_manager.load("DummyObject")
  
  dummy.value = "other_value"
  in_db_2 = db_manager.load("DummyObject")
  
  assert in_db_1["__parent__"] == in_db_2["__parent__"]


def test_i_can_update(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
  
  dummy = DummyObject(session)
  clone = dummy.copy()
  
  clone.number = 34
  clone.value = "other_value"
  clone.other_attr = "some_value"
  
  dummy.update(clone)
  
  assert simplify(db_manager.load("DummyObject")) == {"value": "other_value", "number": 34}


def test_forbidden_attributes_are_not_the_copy(session, db_manager):
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
      
      with self.initializing():
        self.value: str = "hello"
        self.number: int = 42
        self.none_value: None = None
  
  dummy = DummyObject(session)
  
  clone = dummy.copy()
  
  for k in DbObject._forbidden_attrs:
    assert not hasattr(clone, k), f"Clone should not have forbidden attribute '{k}'"


def test_forbidden_attributes_are_not_the_copy_for_dataclass(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
    none_value: None = None
  
  dummy = DummyObject(session)
  
  clone = dummy.copy()
  
  for k in DbObject._forbidden_attrs:
    assert not hasattr(clone, k), f"Clone should not have forbidden attribute '{k}'"


def test_i_cannot_update_a_forbidden_attribute(session, db_manager):
  @dataclass
  class DummyObject(DbObject):
    def __init__(self, sess: dict):
      super().__init__(sess, "DummyObject", db_manager)
    
    value: str = "hello"
    number: int = 42
    none_value: None = None
  
  dummy = DummyObject(session)
  
  dummy.update(_session="other_value")
  
  assert dummy._session == session