wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
406 lines
12 KiB
Python
406 lines
12 KiB
Python
"""Smoke tests for ``application.worker.remote_worker`` and ``sync_worker``.
|
|
|
|
``remote_worker`` in ``sync`` mode does one PG write: it bumps
|
|
``sources.date`` on the referenced source row to ``now()``. That's the
|
|
side-effect we assert here. The remote loader, chunker, embedding
|
|
pipeline, and the backend HTTP callback are all mocked — only the PG
|
|
update is real.
|
|
|
|
``sync_worker`` reads rows out of ``sources`` whose ``sync_frequency``
|
|
matches and dispatches them through ``sync`` → ``remote_worker``. We
|
|
assert one seeded row is discovered and forwarded.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from application.parser.schema.base import Document
|
|
from application.storage.db.repositories.sources import SourcesRepository
|
|
|
|
|
|
@pytest.fixture
|
|
def _mock_remote_pipeline(monkeypatch):
|
|
"""Stub out the non-PG boundaries used by ``remote_worker``."""
|
|
from application import worker
|
|
|
|
fake_loader = MagicMock(name="remote_loader")
|
|
fake_loader.load_data.return_value = [
|
|
Document(
|
|
text="page body",
|
|
extra_info={"file_path": "guides/setup.md", "title": "setup"},
|
|
doc_id="d1",
|
|
)
|
|
]
|
|
monkeypatch.setattr(
|
|
worker.RemoteCreator, "create_loader", lambda loader: fake_loader
|
|
)
|
|
monkeypatch.setattr(
|
|
worker,
|
|
"embed_and_store_documents",
|
|
lambda docs, full_path, source_id, task, **kw: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
worker, "upload_index", lambda full_path, file_data: None
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRemoteWorkerSyncUpdatesDate:
|
|
def test_sync_mode_bumps_source_date(
|
|
self,
|
|
pg_conn,
|
|
patch_worker_db,
|
|
task_self,
|
|
monkeypatch,
|
|
_mock_remote_pipeline,
|
|
):
|
|
from application import worker
|
|
|
|
# Seed a source with a known old ``date`` we can compare against.
|
|
import datetime as dt
|
|
|
|
old_date = dt.datetime(2020, 1, 1, tzinfo=dt.timezone.utc)
|
|
src = SourcesRepository(pg_conn).create(
|
|
"my-remote",
|
|
user_id="bob",
|
|
type="crawler",
|
|
retriever="classic",
|
|
sync_frequency="daily",
|
|
remote_data={"urls": ["http://example.com"]},
|
|
date=old_date,
|
|
)
|
|
source_id = str(src["id"])
|
|
|
|
worker.remote_worker(
|
|
task_self,
|
|
{"urls": ["http://example.com"]},
|
|
"my-remote",
|
|
"bob",
|
|
"crawler",
|
|
directory="temp",
|
|
retriever="classic",
|
|
sync_frequency="daily",
|
|
operation_mode="sync",
|
|
doc_id=source_id,
|
|
)
|
|
|
|
refreshed = SourcesRepository(pg_conn).get(source_id, "bob")
|
|
assert refreshed is not None
|
|
# row_to_dict coerces datetimes to ISO 8601 strings; UTC
|
|
# timezone-aware ISO strings sort chronologically.
|
|
assert refreshed["date"] > old_date.isoformat(), (
|
|
"remote_worker(sync) should push sources.date forward"
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSyncWorker:
|
|
def test_reads_sources_and_dispatches_sync(
|
|
self,
|
|
pg_conn,
|
|
patch_worker_db,
|
|
task_self,
|
|
monkeypatch,
|
|
):
|
|
"""``sync_worker`` selects rows by ``sync_frequency`` and passes
|
|
each to ``sync``. We assert the seeded row is discovered and
|
|
forwarded with the right doc_id — the nested ``sync`` call is
|
|
stubbed so we don't re-run the whole remote pipeline here."""
|
|
from application import worker
|
|
|
|
src = SourcesRepository(pg_conn).create(
|
|
"weekly-feed",
|
|
user_id="carol",
|
|
type="url",
|
|
retriever="classic",
|
|
sync_frequency="weekly",
|
|
remote_data={"url": "http://example.com"},
|
|
)
|
|
|
|
captured: list[dict] = []
|
|
|
|
def _fake_sync(self, source_data, name_job, user, loader,
|
|
sync_frequency, retriever, doc_id=None, directory="temp"):
|
|
captured.append({
|
|
"name_job": name_job,
|
|
"user": user,
|
|
"loader": loader,
|
|
"sync_frequency": sync_frequency,
|
|
"retriever": retriever,
|
|
"doc_id": doc_id,
|
|
})
|
|
return {"status": "success"}
|
|
|
|
monkeypatch.setattr(worker, "sync", _fake_sync)
|
|
|
|
result = worker.sync_worker(task_self, "weekly")
|
|
|
|
assert result["total_sync_count"] == 1
|
|
assert result["sync_success"] == 1
|
|
assert len(captured) == 1
|
|
assert captured[0]["name_job"] == "weekly-feed"
|
|
assert captured[0]["user"] == "carol"
|
|
assert captured[0]["loader"] == "url"
|
|
assert captured[0]["doc_id"] == str(src["id"])
|
|
|
|
def test_connector_sources_are_skipped(
|
|
self,
|
|
pg_conn,
|
|
patch_worker_db,
|
|
task_self,
|
|
monkeypatch,
|
|
):
|
|
"""connector:* sources have no RemoteCreator loader — sync_worker
|
|
must skip them, not dispatch them into sync()."""
|
|
from application import worker
|
|
|
|
SourcesRepository(pg_conn).create(
|
|
"drive-folder",
|
|
user_id="dave",
|
|
type="connector:file",
|
|
retriever="classic",
|
|
sync_frequency="daily",
|
|
remote_data={
|
|
"provider": "google_drive",
|
|
"file_ids": ["f1"],
|
|
"folder_ids": [],
|
|
"recursive": False,
|
|
},
|
|
)
|
|
|
|
def _must_not_run(*args, **kwargs):
|
|
raise AssertionError("sync() must not run for connector sources")
|
|
|
|
monkeypatch.setattr(worker, "sync", _must_not_run)
|
|
|
|
result = worker.sync_worker(task_self, "daily")
|
|
|
|
assert result["total_sync_count"] == 1
|
|
assert result["sync_skipped"] == 1
|
|
assert result["sync_success"] == 0
|
|
assert result["sync_failure"] == 0
|
|
|
|
def test_dict_remote_data_is_normalized_before_loader(
|
|
self,
|
|
pg_conn,
|
|
patch_worker_db,
|
|
task_self,
|
|
monkeypatch,
|
|
):
|
|
"""Regression: remote_data reads back as a dict; sync_worker must
|
|
hand the loader the URL string, not the raw dict."""
|
|
from application import worker
|
|
|
|
SourcesRepository(pg_conn).create(
|
|
"docs-crawl",
|
|
user_id="erin",
|
|
type="crawler",
|
|
retriever="classic",
|
|
sync_frequency="weekly",
|
|
remote_data={"url": "https://example.com", "provider": "crawler"},
|
|
)
|
|
|
|
received: list = []
|
|
fake_loader = MagicMock(name="remote_loader")
|
|
|
|
def _capture(source_data):
|
|
received.append(source_data)
|
|
return [
|
|
Document(
|
|
text="page body",
|
|
extra_info={"file_path": "index.md", "title": "home"},
|
|
doc_id="d1",
|
|
)
|
|
]
|
|
|
|
fake_loader.load_data.side_effect = _capture
|
|
monkeypatch.setattr(
|
|
worker.RemoteCreator, "create_loader", lambda loader: fake_loader
|
|
)
|
|
monkeypatch.setattr(
|
|
worker,
|
|
"embed_and_store_documents",
|
|
lambda docs, full_path, source_id, task, **kw: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
worker, "upload_index", lambda full_path, file_data: None
|
|
)
|
|
|
|
result = worker.sync_worker(task_self, "weekly")
|
|
|
|
assert result["total_sync_count"] == 1
|
|
assert result["sync_success"] == 1
|
|
assert result["sync_failure"] == 0
|
|
assert received == ["https://example.com"], (
|
|
"loader must receive the URL string, not the remote_data dict"
|
|
)
|
|
|
|
def test_unsyncable_remote_data_is_skipped(
|
|
self,
|
|
pg_conn,
|
|
patch_worker_db,
|
|
task_self,
|
|
monkeypatch,
|
|
):
|
|
"""A URL source whose remote_data dict has no URL key normalizes
|
|
to None — sync_worker must skip it, not dispatch a doomed sync()."""
|
|
from application import worker
|
|
|
|
SourcesRepository(pg_conn).create(
|
|
"broken-feed",
|
|
user_id="frank",
|
|
type="url",
|
|
retriever="classic",
|
|
sync_frequency="monthly",
|
|
remote_data={"provider": "url"},
|
|
)
|
|
|
|
def _must_not_run(*args, **kwargs):
|
|
raise AssertionError("sync() must not run for unsyncable sources")
|
|
|
|
monkeypatch.setattr(worker, "sync", _must_not_run)
|
|
|
|
result = worker.sync_worker(task_self, "monthly")
|
|
|
|
assert result["total_sync_count"] == 1
|
|
assert result["sync_skipped"] == 1
|
|
assert result["sync_failure"] == 0
|
|
assert result["sync_success"] == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRemoteWorkerPathTraversal:
|
|
"""Regression: ``name_job`` must not be usable as a path segment.
|
|
|
|
Historically ``remote_worker`` built its workspace from
|
|
``os.path.join(directory, user, name_job)`` and cleaned it up with
|
|
``shutil.rmtree`` in a ``finally``. A ``name_job`` like
|
|
``../../evil`` would therefore let an authenticated caller delete
|
|
directories outside the intended ``<directory>/<user>/`` root.
|
|
The fix uses a random uuid leaf; ``name_job`` is metadata only.
|
|
"""
|
|
|
|
def test_traversal_name_job_does_not_escape_user_workspace(
|
|
self,
|
|
tmp_path,
|
|
task_self,
|
|
monkeypatch,
|
|
_mock_remote_pipeline,
|
|
):
|
|
from application import worker
|
|
|
|
created_paths: list[str] = []
|
|
deleted_paths: list[str] = []
|
|
real_makedirs = os.makedirs
|
|
real_rmtree = worker.shutil.rmtree
|
|
|
|
def _spy_makedirs(path, *args, **kwargs):
|
|
created_paths.append(path)
|
|
return real_makedirs(path, *args, **kwargs)
|
|
|
|
def _spy_rmtree(path, *args, **kwargs):
|
|
deleted_paths.append(path)
|
|
return real_rmtree(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(worker.os, "makedirs", _spy_makedirs)
|
|
monkeypatch.setattr(worker.shutil, "rmtree", _spy_rmtree)
|
|
|
|
directory = str(tmp_path / "temp")
|
|
user = "bob"
|
|
malicious_name = "../../evil"
|
|
|
|
worker.remote_worker(
|
|
task_self,
|
|
{"urls": ["http://example.com"]},
|
|
malicious_name,
|
|
user,
|
|
"crawler",
|
|
directory=directory,
|
|
operation_mode="upload",
|
|
)
|
|
|
|
directory_real = os.path.realpath(directory)
|
|
user_root = os.path.realpath(os.path.join(directory, user))
|
|
|
|
rmtree_targets = [
|
|
os.path.realpath(p)
|
|
for p in deleted_paths
|
|
if os.path.realpath(p).startswith(directory_real)
|
|
]
|
|
assert len(rmtree_targets) == 1, rmtree_targets
|
|
assert rmtree_targets[0].startswith(user_root + os.sep), (
|
|
f"rmtree target {rmtree_targets[0]} escaped {user_root}"
|
|
)
|
|
assert malicious_name not in "".join(created_paths + deleted_paths)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRemoteWorkerDeterministicSourceId:
|
|
"""Upload-mode ``remote_worker`` derives a stable ``source_id`` from the key."""
|
|
|
|
def test_uses_uuid5_when_idempotency_key_present(
|
|
self,
|
|
tmp_path,
|
|
task_self,
|
|
monkeypatch,
|
|
_mock_remote_pipeline,
|
|
):
|
|
from application import worker
|
|
|
|
captured: list[dict] = []
|
|
monkeypatch.setattr(
|
|
worker, "upload_index",
|
|
lambda full_path, file_data: captured.append(file_data),
|
|
)
|
|
|
|
for _ in range(2):
|
|
worker.remote_worker(
|
|
task_self,
|
|
{"urls": ["http://example.com"]},
|
|
"feed",
|
|
"bob",
|
|
"crawler",
|
|
directory=str(tmp_path / "temp"),
|
|
operation_mode="upload",
|
|
idempotency_key="abc",
|
|
)
|
|
|
|
expected = str(uuid.uuid5(worker.DOCSGPT_INGEST_NAMESPACE, "abc"))
|
|
assert len(captured) == 2
|
|
assert captured[0]["id"] == expected
|
|
assert captured[1]["id"] == expected
|
|
|
|
def test_falls_back_to_uuid4_without_key(
|
|
self,
|
|
tmp_path,
|
|
task_self,
|
|
monkeypatch,
|
|
_mock_remote_pipeline,
|
|
):
|
|
from application import worker
|
|
|
|
captured: list[dict] = []
|
|
monkeypatch.setattr(
|
|
worker, "upload_index",
|
|
lambda full_path, file_data: captured.append(file_data),
|
|
)
|
|
|
|
for _ in range(2):
|
|
worker.remote_worker(
|
|
task_self,
|
|
{"urls": ["http://example.com"]},
|
|
"feed",
|
|
"bob",
|
|
"crawler",
|
|
directory=str(tmp_path / "temp"),
|
|
operation_mode="upload",
|
|
)
|
|
|
|
assert len(captured) == 2
|
|
assert captured[0]["id"] != captured[1]["id"]
|