the holy quest for the holy home
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

64 lines
3.0 KiB

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 = """
<html><head><meta property="og:title" content="Casa 3 quartos"></head><body>
<img src="https://img.dfimoveis.com.br/fotos/1401074/a.webp">
<img src="https://img.dfimoveis.com.br/fotos/999999/b.webp">
</body></html>
"""
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 = '<meta property="og:title" content="Casa"><img src="https://img.dfimoveis.com.br/fotos/123/a.webp">'
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()