Feat: Aplicação de testes unitários

.github/workflows/tests.yml

@@ -0,0 +1,31 @@
+name: Pytest no PR
+
+on:
+  pull_request:
+    branches: [ "master" ] 
+
+jobs:
+  testes:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout do código
+      uses: actions/checkout@v4 
+
+    - name: Configurar Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.13.3' 
+
+    - name: Instalar dependências
+      run: |
+        python -m pip install --upgrade pip
+        pip install pytest
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+
+    - name: Executar Pytest
+      run: |
+        mkdir -p data
+        touch data/.gitkeep
+        export PYTHONPATH=$PYTHONPATH:$(pwd)
+        pytest

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

core/buffer_manager.py

@@ -1,51 +1,74 @@
 # Decide se vai usar cache ou disco
 from core.page_manager import PageManager
 
-class BufferManager():
+
+class BufferManager:
     def __init__(self, page_manager, max_pages=10):
-        self.page_manager = page_manager 
-        self.max_pages = max_pages 
-        self.pages = {} # cache
-        self.order = [] # controla FIFO
+        self.page_manager = page_manager
+        self.max_pages = max_pages
+        self.pages = {}  # cache
+        self.order = []  # controla FIFO
 
-    def get_page(self, page_id): # devolve buffer
+    def get_page(self, page_id):  # devolve buffer
         # se ja está no buffer
         print(f"BUFFER_MANAGER: get_page {page_id}")
+
         if page_id in self.pages:
             entry = self.pages[page_id]
             entry["pin"] += 1
-            return entry["buffer"] 
-        
+            return entry["buffer"]
+
         # buffer cheio -> precisa expulsar alguem
-        if len(self.pages) > self.max_pages:
+        if len(self.pages) >= self.max_pages:
             self._evict()
 
         # lê do disco, coloca no buffer e marca como em uso
         buffer = self.page_manager.load_page(page_id)
 
+        self.pages[page_id] = {"buffer": buffer, "dirty": False, "pin": 1}
+
+        self.order.append(page_id)
+        return buffer
+
+    def create_page(self, page_type="heap"):
+        print("BUFFER_MANAGER: create_page")
+        if len(self.pages) >= self.max_pages:
+            self._evict()
+
+        page_id = self.page_manager.create_page(page_type)
+
+        buffer = self.page_manager.load_page(page_id)
+
         self.pages[page_id] = {
             "buffer": buffer,
-            "dirty": False,
-            "pin": 1
+            "dirty": True,  # já começa suja
+            "pin": 1,  # já em uso
         }
 
         self.order.append(page_id)
-        return buffer
 
-    def mark_dirty(self, page_id): # marca página como modified
+        return page_id, buffer
+
+    def mark_dirty(self, page_id):  # marca página como modified
         print(f"BUFFER_MANAGER: mark_dirty {page_id}")
         self.pages[page_id]["dirty"] = True
 
     def unpin(self, page_id):
         print(f"BUFFER_MANAGER: unpin {page_id}")
-        self.pages[page_id]["pin"] -= 1
 
-    #FIFO
+        entry = self.pages[page_id]
+
+        if entry["pin"] <= 0:
+            raise RuntimeError(f"Página {page_id} já está despinada")
+
+        entry["pin"] -= 1
+
+    # FIFO
     # remove da ram caso esteja cheia
     # dirty = suja
     def _evict(self):
         print("BUFFER_MANAGER: _evict")
-        for page_id in self.order:
+        for page_id in list(self.order):
             entry = self.pages[page_id]
 
             if entry["pin"] > 0:
@@ -60,10 +83,9 @@ def _evict(self):
 
         raise RuntimeError("Todas as páginas estão pinadas")
 
-    def flush_all(self): # grava tudo
+    def flush_all(self):  # grava tudo
         print("BUFFER_MANAGER: flush_Fall")
         for page_id, entry in self.pages.items():
             if entry["dirty"]:
                 self.page_manager.save_page(entry["buffer"], page_id)
                 entry["dirty"] = False
-

core/heap_manager.py

@@ -1,35 +1,43 @@
 # Só pede página
-from core.page_manager import PageManager
 from core.page import insert_record, iter_records
 
 
 class HeapManager:
-    def __init__(self, page_manager):
-        self.page_manager = page_manager
+    def __init__(self, buffer_manager):
+        self.buffer_manager = buffer_manager
         self.heap_pages = []  # lista de page_ids
 
     def insert(self, record_bytes):
         print("HEAP MANAGER: insert")
         for page_id in self.heap_pages:
-            buffer = self.page_manager.load_page(page_id)
+            buffer = self.buffer_manager.get_page(page_id)
+
             try:
                 offset = insert_record(buffer, record_bytes)
-                self.page_manager.save_page(buffer, page_id)
-                return page_id, offset
             except ValueError:
-                continue
+                pass
+            else:
+                self.buffer_manager.mark_dirty(page_id)
+                return page_id, offset
+            finally:
+                self.buffer_manager.unpin(page_id)
 
-        page_id, buffer = self.page_manager.create_page("heap")
-        offset = insert_record(buffer, record_bytes)
+        page_id, buffer = self.buffer_manager.create_page("heap")
 
-        self.page_manager.save_page(buffer, page_id)
         self.heap_pages.append(page_id)
 
+        offset = insert_record(buffer, record_bytes)
+        self.buffer_manager.mark_dirty(page_id)
+        self.buffer_manager.unpin(page_id)
+
         return page_id, offset
 
     def scan(self):
         print("HEAP MANAGER: scan")
         for page_id in self.heap_pages:
-            buffer = self.page_manager.load_page(page_id)
-            for record in iter_records(buffer):
-                yield record
+            buffer = self.buffer_manager.get_page(page_id)
+            try:
+                for record in iter_records(buffer):
+                    yield record
+            finally:
+                self.buffer_manager.unpin(page_id)

core/page.py

@@ -1,6 +1,6 @@
 from core.config import PAGE_SIZE, DB_MAGIC, DB_VERSION
 
-mapping = {"heap": 1, "index": 2}
+mapping = {"heap": 1, "index": 2}  # necessita de ajuste
 mutable_fields = ["flags", "free_start", "free_end", "lsn"]
 
 
@@ -69,7 +69,7 @@ def validate_header(PageHeader):
         return True
 
 
-# atualiza campos imutáveis
+# atualiza campos mutáveis
 def update_header_field(PageHeader, buffer, updates):
     if "free_start" in updates or "free_end" in updates:
         new_free_start = updates.get(
@@ -116,7 +116,7 @@ def insert_record(buffer, record_bytes):
 
 
 def read_record(buffer, offset):
-    record_size = int.from_bytes(buffer[offset : offset + 2], byteorder="little")
+    record_size = int.from_bytes(buffer[offset + 1 : offset + 3], byteorder="little")
     record_data = buffer[offset + 2 : offset + 2 + record_size]
 
     return record_data

core/page_manager.py

@@ -8,12 +8,11 @@
 class PageManager:
     def __init__(self, data_dir=DATA_DIR):
         self.data_dir = data_dir
-        self.pages = {}  # page_id -> buffer
         os.makedirs(self.data_dir, exist_ok=True)
-        self.next_page_id = self._diskover_next_page_id()
+        self.next_page_id = self._dicover_next_page_id()
 
-    def _diskover_next_page_id(self):
-        print("PAGE MANAGER: _diskover_next_page_id")
+    def _dicover_next_page_id(self):
+        print("PAGE MANAGER: _dicover_next_page_id")
         files = os.listdir(self.data_dir)
         page_ids = []
 
@@ -37,7 +36,7 @@ def create_page(self, page_type="heap"):
         self.write_page_to_disk(buffer, page_id)
 
         self.next_page_id += 1
-        return page_id, buffer
+        return page_id
 
     def load_page(self, page_id):
         print(f"PAGE MANAGER: load_page {page_id}")

core/table_manager.py

@@ -2,14 +2,14 @@
 
 
 class TableManager:
-    def __init__(self, page_manager, catalog):
-        self.page_manager = page_manager
+    def __init__(self, buffer_manager, catalog):
+        self.buffer_manager = buffer_manager
         self.catalog = catalog
         self.heaps = {}
 
     def create_table(self, name):
         self.catalog.create_table(name)
-        heap = HeapManager(self.page_manager)
+        heap = HeapManager(self.buffer_manager)
 
         table_meta = self.catalog.get_table(name)
         heap.heap_pages = table_meta["heap_pages"]

tests/test_buffermanager.py

@@ -0,0 +1,30 @@
+import pytest
+import os
+from core.page_manager import PageManager
+from core.buffer_manager import BufferManager
+from core.config import DATA_DIR
+
+
+@pytest.fixture
+def clean_data_dir():
+    for f in os.listdir(DATA_DIR):
+        if f.startswith("page_"):
+            os.remove(os.path.join(DATA_DIR, f))
+
+
+def test_buffer_create_get_flush(clean_data_dir):
+    pm = PageManager()
+    bm = BufferManager(pm, max_pages=2)
+    page_id, buffer = bm.create_page("heap")
+
+    # modifica o buffer
+    buffer[64] = 123
+    bm.mark_dirty(page_id)
+    bm.unpin(page_id)
+
+    bm.flush_all()
+
+    # recarrega
+    loaded = bm.get_page(page_id)
+    assert loaded[64] == 123
+    bm.unpin(page_id)

tests/test_catalog.py

@@ -0,0 +1,22 @@
+import pytest
+import os
+from core.catalog import Catalog
+from core.config import DATA_DIR
+
+CATALOG_FILE = os.path.join(DATA_DIR, "catalog.json")
+
+
+@pytest.fixture
+def clean_catalog():
+    if os.path.exists(CATALOG_FILE):
+        os.remove(CATALOG_FILE)
+
+
+def test_create_table_add_get(clean_catalog):
+    cat = Catalog()
+    cat.create_table("users")
+    cat.add_heap_page("users", 0)
+
+    table = cat.get_table("users")
+    assert table is not None
+    assert 0 in table["heap_pages"]

tests/test_heap_manager.py

@@ -0,0 +1,29 @@
+import pytest
+import os
+from core.page_manager import PageManager
+from core.buffer_manager import BufferManager
+from core.heap_manager import HeapManager
+from core.config import DATA_DIR
+
+
+@pytest.fixture
+def clean_data_dir():
+    for f in os.listdir(DATA_DIR):
+        if f.startswith("page_"):
+            os.remove(os.path.join(DATA_DIR, f))
+
+
+def test_heap_insert_scan(clean_data_dir):
+    pm = PageManager()
+    bm = BufferManager(pm)
+    heap = HeapManager(bm)
+
+    r1 = b"record1"
+    r2 = b"record2"
+
+    pid1, off1 = heap.insert(r1)
+    pid2, off2 = heap.insert(r2)
+
+    records = list(heap.scan())
+    assert r1 in records
+    assert r2 in records

tests/test_page.py

@@ -0,0 +1,17 @@
+from core.page import create_empty_page, insert_record, iter_records
+
+
+def test_create_empty_page():
+    buffer = create_empty_page("heap", 0)
+    # Checa header básico
+    assert buffer[6:8] == (1).to_bytes(2, "little")  # heap
+    assert len(buffer) == 8192
+
+
+def test_insert_and_iter_record():
+    buffer = create_empty_page("heap", 0)
+    record = b"hello"
+    offset = insert_record(buffer, record)
+    records = list(iter_records(buffer))
+    assert record in records
+    assert offset >= 0