import tempfile
import unittest
from pathlib import Path
from dfimoveis_scraper import Database, Parser, normalize_creci, normalize_listing_address
class SchemaV2Tests(unittest.TestCase):
def test_normalization(self):
self.assertEqual(normalize_creci("26.903 Ver mais qualquer texto"), "26903")
self.assertEqual(
normalize_listing_address(
"https://www.dfimoveis.com.br/imovel/casa-venda-jardins-mangueiral-qc-13-rua-g-123",
"texto contaminado " * 100,
),
"Jardins Mangueiral - QC 13 - Rua G",
)
def test_parser_rejects_recommendation_photos(self):
page = """
"""
data, photos = Parser.detail(
"https://www.dfimoveis.com.br/imovel/casa-3-quartos-venda-jardins-mangueiral-qc-13-1401074",
page,
)
self.assertEqual(photos, ["https://img.dfimoveis.com.br/fotos/1401074/a.webp"])
self.assertEqual(data["creci_normalized"], None)
self.assertEqual(data["address_normalized"], "Jardins Mangueiral - QC 13")
def test_schema_migration_is_additive(self):
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "test.sqlite3"
db = Database(path)
columns = {row[1] for row in db.conn.execute("PRAGMA table_info(listings)")}
self.assertIn("stable_content_hash", columns)
self.assertIn("creci_normalized", columns)
self.assertTrue(db.conn.execute("SELECT 1 FROM sqlite_master WHERE name='data_quality_issues'").fetchone())
db.close()
def test_raw_ui_noise_does_not_create_material_change(self):
page = '
'
url = "https://www.dfimoveis.com.br/imovel/casa-venda-jardins-mangueiral-qc-13-123"
data, _ = Parser.detail(url, page)
with tempfile.TemporaryDirectory() as tmp:
db = Database(Path(tmp) / "test.sqlite3")
run_id = db.start_run([url], {"test": True})
self.assertTrue(db.upsert_listing(data, url, run_id))
data["creci"] = "texto instável da interface"
data["address"] = "outro bloco contaminado"
data["raw_content_hash"] = "different-raw-hash"
self.assertFalse(db.upsert_listing(data, url, run_id))
self.assertEqual(db.conn.execute("SELECT count(*) FROM listing_history").fetchone()[0], 1)
self.assertEqual(db.conn.execute("SELECT count(*) FROM listing_change_events").fetchone()[0], 1)
self.assertEqual(db.mark_inactive(url, set(), "9999-01-01T00:00:00+00:00"), 1)
self.assertEqual(db.mark_inactive(url, set(), "9999-01-01T00:00:00+00:00"), 0)
db.close()
if __name__ == "__main__":
unittest.main()