Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .flake8
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[flake8]
max-line-length = 200
ignore = E501,E203,W503,E402
ignore = E501,E203,W503,E402,F401
exclude =
.git,
__pycache__,
Expand Down
3 changes: 2 additions & 1 deletion requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -7,4 +7,5 @@ coverage
flake8==6.1.0
black==25.9.0
scikit-learn
dotenv
dotenv
numpy
100 changes: 100 additions & 0 deletions tests/test_data_loader.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
import pytest
from pathlib import Path
from src.preprocessing.data_loader import load_processed_text
import subprocess
import shutil


def test_load_processed_text_reads_files(tmp_path):
data_dir = tmp_path / "data" / "processed-text"
data_dir.mkdir(parents=True)

# Create mock text files
file1 = data_dir / "file1.txt"
file2 = data_dir / "file2.txt"
file1.write_text("Predator diet data example 1", encoding="utf-8")
file2.write_text("Predator diet data example 2", encoding="utf-8")

texts = load_processed_text(str(data_dir))

assert len(texts) == 2
assert "example 1" in texts[0]
assert "example 2" in texts[1]


def test_load_processed_text_empty_directory(tmp_path):
empty_dir = tmp_path / "data" / "processed-text"
empty_dir.mkdir(parents=True)

texts = load_processed_text(str(empty_dir))

assert isinstance(texts, list)
assert len(texts) == 0


def test_load_processed_text_ignores_non_txt(tmp_path):
data_dir = tmp_path / "data" / "processed-text"
data_dir.mkdir(parents=True)

(data_dir / "ignore.pdf").write_text("Not a text file", encoding="utf-8")
(data_dir / "keep.txt").write_text("Keep this one", encoding="utf-8")

texts = load_processed_text(str(data_dir))

assert len(texts) == 1
assert texts[0] == "Keep this one"


def test_load_processed_text_encoding(tmp_path):
data_dir = tmp_path / "data" / "processed-text"
data_dir.mkdir(parents=True)

content = "Café with UTF-8 accents"
file_path = data_dir / "utf8_test.txt"
file_path.write_text(content, encoding="utf-8")

texts = load_processed_text(str(data_dir))

assert len(texts) == 1
assert "Café" in texts[0]


def test_main_prints_summary(monkeypatch, tmp_path, capsys):
from src.preprocessing import data_loader

data_dir = tmp_path / "data" / "processed-text"
data_dir.mkdir(parents=True)
(data_dir / "example.txt").write_text("Sample content", encoding="utf-8")

monkeypatch.chdir(tmp_path)
data_loader.texts = data_loader.load_processed_text(str(data_dir))

data_loader.__main__ = True
print(f"Loaded {len(data_loader.texts)} text files.")
if data_loader.texts:
print(f"First file preview:\n{data_loader.texts[0][:300]}...")

captured = capsys.readouterr()
assert "Loaded 1 text files." in captured.out
assert "Sample content" in captured.out


def test_data_loader_main_executes(tmp_path):
data_dir = tmp_path / "data" / "processed-text"
data_dir.mkdir(parents=True)
(data_dir / "sample.txt").write_text("Predator diet study data", encoding="utf-8")

script_path = Path("src/preprocessing/data_loader.py")
tmp_script_path = tmp_path / "data_loader.py"
shutil.copy(script_path, tmp_script_path)

result = subprocess.run(
["python", str(tmp_script_path)],
cwd=tmp_path,
capture_output=True,
text=True,
)

assert result.returncode == 0
assert "Loaded 1 text files." in result.stdout
assert "Predator diet study data" in result.stdout
102 changes: 102 additions & 0 deletions tests/test_generate_labels.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
import pytest
import json
import subprocess
from pathlib import Path
from src.preprocessing.generate_labels import generate_labels


def test_generate_labels_creates_json(tmp_path):
useful_dir = tmp_path / "data" / "useful"
not_useful_dir = tmp_path / "data" / "not-useful"
output_file = tmp_path / "data" / "labels.json"

# Create mock PDF files
useful_dir.mkdir(parents=True)
not_useful_dir.mkdir(parents=True)
(useful_dir / "useful.pdf").touch()
(not_useful_dir / "not-useful.pdf").touch()

# Run function
generate_labels(useful_dir, not_useful_dir, output_file)

# Assertions
assert output_file.exists(), "labels.json should be created"
with open(output_file, "r", encoding="utf-8") as f:
labels = json.load(f)

assert "useful.txt" in labels
assert "not-useful.txt" in labels
assert labels["useful.txt"] == "useful"
assert labels["not-useful.txt"] == "not useful"
assert len(labels) == 2


def test_generate_labels_empty_folders(tmp_path):
useful_dir = tmp_path / "data" / "useful"
not_useful_dir = tmp_path / "data" / "not-useful"
output_file = tmp_path / "data" / "labels.json"

useful_dir.mkdir(parents=True)
not_useful_dir.mkdir(parents=True)

generate_labels(useful_dir, not_useful_dir, output_file)

with open(output_file, "r", encoding="utf-8") as f:
labels = json.load(f)

assert labels == {}, "Empty folders should produce an empty JSON"


def test_generate_labels_overwrites_existing(tmp_path):
output_file = tmp_path / "data" / "labels.json"
output_file.parent.mkdir(parents=True)
with open(output_file, "w", encoding="utf-8") as f:
json.dump({"old_file.txt": "useful"}, f)

useful_dir = tmp_path / "data" / "useful"
not_useful_dir = tmp_path / "data" / "not-useful"
useful_dir.mkdir(parents=True)
not_useful_dir.mkdir(parents=True)
(useful_dir / "new_file.pdf").touch()

generate_labels(useful_dir, not_useful_dir, output_file)

with open(output_file, "r", encoding="utf-8") as f:
labels = json.load(f)

assert "old_file.txt" not in labels
assert "new_file.txt" in labels
assert labels["new_file.txt"] == "useful"


def test_generate_labels_cli(tmp_path):
data_dir = tmp_path / "data"
useful_dir = data_dir / "useful"
not_useful_dir = data_dir / "not-useful"
useful_dir.mkdir(parents=True)
not_useful_dir.mkdir(parents=True)
(useful_dir / "CLI_Test.pdf").touch()
(not_useful_dir / "CLI_Not.pdf").touch()

import shutil

repo_src = Path("src")
tmp_src = tmp_path / "src"
shutil.copytree(repo_src, tmp_src)

result = subprocess.run(
["python", "src/preprocessing/generate_labels.py"],
capture_output=True,
text=True,
cwd=tmp_path,
)

assert result.returncode == 0, f"CLI failed with error:\n{result.stderr}"
assert "labels.json created" in result.stdout

output_file = tmp_path / "data" / "labels.json"
assert output_file.exists(), "labels.json should be created in tmp_path/data"

labels = json.loads(output_file.read_text(encoding="utf-8"))
assert "CLI_Test.txt" in labels
assert "CLI_Not.txt" in labels
82 changes: 82 additions & 0 deletions tests/test_pdf_classifier.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
import pytest
import joblib
from pathlib import Path
from sklearn.linear_model import LogisticRegression
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.feature_extraction.text import CountVectorizer
from unittest.mock import patch, MagicMock
from src.model.pdf_classifier import classify_pdf


@pytest.fixture
def model_dir_with_mock_model(tmp_path):
model_dir = tmp_path / "models"
model_dir.mkdir(parents=True)

texts = ["predator stomach content", "fish prey analysis", "rock study geology", "mineral content paper"]
labels = ["useful", "useful", "not useful", "not useful"]

vectorizer = TfidfVectorizer(max_features=4)
X = vectorizer.fit_transform(texts)
model = LogisticRegression().fit(X, labels)

joblib.dump(model, model_dir / "pdf_classifier_model.pkl")
joblib.dump(vectorizer, model_dir / "tfidf_vectorizer.pkl")

return model_dir


@patch("src.model.pdf_classifier.extract_text_from_pdf", return_value="predator stomach content analysis")
def test_classify_pdf_valid_case(mock_extract, model_dir_with_mock_model, capsys):
test_pdf = Path("tests/test.pdf")
classify_pdf(test_pdf, model_dir_with_mock_model)

output = capsys.readouterr().out
assert "PDF Classification Result" in output
assert "useful" in output.lower()
assert "%" in output


def test_classify_pdf_missing_model(capsys, tmp_path):
model_dir = tmp_path / "empty_models"
model_dir.mkdir()

test_pdf = Path("tests/test.pdf")
classify_pdf(test_pdf, model_dir)

output = capsys.readouterr().out
assert "[ERROR]" in output
assert "Missing model" in output


@patch("src.model.pdf_classifier.extract_text_from_pdf", return_value="")
def test_classify_pdf_no_text(mock_extract, model_dir_with_mock_model, capsys):
test_pdf = Path("tests/empty.pdf")
classify_pdf(test_pdf, model_dir_with_mock_model)

output = capsys.readouterr().out
assert "[ERROR]" in output
assert "No text extracted" in output


@patch("src.model.pdf_classifier.extract_text_from_pdf", return_value="diet prey stomach analysis")
def test_classify_pdf_confidence_and_prediction(mock_extract, model_dir_with_mock_model, capsys):
test_pdf = Path("tests/valid.pdf")
classify_pdf(test_pdf, model_dir_with_mock_model)

output = capsys.readouterr().out
assert "Prediction:" in output
assert "Confidence" not in output
assert "useful" in output.lower()


def test_classify_pdf_handles_nonexistent_model_dir(tmp_path, capsys):
model_dir = tmp_path / "nonexistent_dir"
test_pdf = Path("tests/missing_model.pdf")

classify_pdf(test_pdf, model_dir)

output = capsys.readouterr().out
assert "[ERROR]" in output
assert "Missing model" in output
Loading