diff --git a/README.md b/README.md index ed5d93695e0..595478706db 100644 --- a/README.md +++ b/README.md @@ -97,14 +97,15 @@ Submit the job and retrieve results: ds.submit_python_job( user="do@org.com", code_path="analysis.py", + job_name="analysis", ) ds.sync(); do.sync() # Data owner Approves & runs job -do.jobs[0].approve() +do.jobs["ds@org.com"]["analysis"].approve() do.process_approved_jobs(share_outputs_with_submitter=True) do.sync(); ds.sync() -result = open(ds.jobs[-1].output_paths[0]).read() +result = open(ds.jobs["do@org.com"]["analysis"].output_paths[0]).read() ``` ## Packages diff --git a/docs/API.md b/docs/API.md index 32b0ab2841d..3e34b96d729 100644 --- a/docs/API.md +++ b/docs/API.md @@ -59,7 +59,43 @@ Returns a `PeerList`. Get the list of jobs. Auto-syncs before returning. -Returns a `JobsList`. +Returns a `JobsList`. Address a job by **email, then name** — a job name is +unique per datasite and submitter, so the email is what makes the name resolve +to one job, and both parts stay the same as jobs are added: + +```python +# as the data scientist, naming the data owner +client.jobs["do@org.com"]["analysis"].output_paths + +# as the data owner, naming the submitter +client.jobs["ds@org.com"]["analysis"].approve() +``` + +**An email keeps the jobs it is a party to**, on either side: the datasite they +sit on, or the person who submitted them. Usually that is the other party — a +data scientist names the data owner, a data owner names the submitter — but +naming yourself works and keeps your own, which is what a `PermissionError` on +someone else's job suggests. Job names cannot contain `@`, so the two kinds of +key never collide. + +Chain both emails when one submitter sent the same name to two datasites: + +```python +client.jobs["do@org.com"]["ds@org.com"]["analysis"].approve() +``` + +A bare name (`client.jobs["analysis"]`) searches every datasite at once. It +still works, but it raises when more than one job answers to it, and the message +names whichever key separates them. + +A job name can no longer hold an `@`. A job submitted before that rule still +resolves by name, with a `DeprecationWarning`; the next version will not resolve +it, so rename such a job. + +Positional indexing (`client.jobs[0]`) also works and matches the Index column +in the table, but positions shift as jobs are added, so prefer an email and a +name. A position is worth using in one case: two jobs that share a datasite, a +submitter and a name, which no email separates. ### `client.datasets` @@ -174,6 +210,7 @@ Submit a Python job to a Data Owner. **DS only.** ds_client.submit_python_job( user="owner@example.com", code_path="/path/to/script.py", + job_name="analysis", ) ``` @@ -195,11 +232,23 @@ Run all approved jobs. **DO only.** - `stream_output`: Stream stdout/stderr in real-time. - `timeout`: Timeout in seconds per job (default: 300). - `force_execution`: Skip version compatibility checks. +- `ignore_peer_version`: Run jobs from peers whose version is incompatible. ```python do_client.process_approved_jobs() ``` +A job whose submitter runs an incompatible version is not run. Each one is +reported by name, with the submitter and the reason: + +``` +⏭️ 1 approved job(s) did not run: + • analysis (submitted by ds@example.com): Skipping peer ds@example.com: incompatible version. + Pass ignore_peer_version=True to run them anyway. +``` + +The job stays at `approved`, so it runs on the next call once the versions match. + --- ## Cleanup diff --git a/packages/syft-bg/src/syft_bg/__init__.py b/packages/syft-bg/src/syft_bg/__init__.py index 89e5239ffa2..e293ece76b9 100644 --- a/packages/syft-bg/src/syft_bg/__init__.py +++ b/packages/syft-bg/src/syft_bg/__init__.py @@ -1,5 +1,7 @@ __version__ = "0.2.2" +from syft_job.logging_config import configure_package_logger + from syft_bg.api import ( AuthResult, AutoApproveResult, @@ -61,3 +63,7 @@ def __getattr__(name: str): print("No config file found, run syft_bg.init() first") return raise AttributeError(f"module 'syft_bg' has no attribute {name!r}") + + +# Last, so the imports stay at the top. Nothing here logs at import time. +configure_package_logger(__name__) diff --git a/packages/syft-enclave/pyproject.toml b/packages/syft-enclave/pyproject.toml index dea2d25aabf..12f05306279 100644 --- a/packages/syft-enclave/pyproject.toml +++ b/packages/syft-enclave/pyproject.toml @@ -10,6 +10,7 @@ requires-python = ">=3.10" dependencies = [ "syft>=0.10.0", # floor: the `syft` PyPI project also hosts legacy PySyft <=0.9 "syft-rds>=0.6.1", + "syft-job==0.1.40", # imported directly for logging_config "pydantic-settings>=2.11.0", "requests>=2.32.0", "google-auth[pyjwt]>=2.22.0", @@ -22,6 +23,7 @@ build-backend = "hatchling.build" [tool.uv.sources] "syft" = { workspace = true } "syft-rds" = { workspace = true } +"syft-job" = { workspace = true } [tool.hatch.build.targets.wheel] packages = ["src/syft_enclaves"] diff --git a/packages/syft-enclave/src/syft_enclaves/__init__.py b/packages/syft-enclave/src/syft_enclaves/__init__.py index c7c5e509ef7..35855ad8b98 100644 --- a/packages/syft-enclave/src/syft_enclaves/__init__.py +++ b/packages/syft-enclave/src/syft_enclaves/__init__.py @@ -1,3 +1,4 @@ +from syft_job.logging_config import configure_package_logger from syft_enclaves.client import SyftEnclaveClient from syft_enclaves.login import login_do, login_ds from syft_enclaves.runner import EnclaveRunner @@ -10,3 +11,6 @@ "login_do", "login_ds", ] + +# Last, so the imports stay at the top. Nothing here logs at import time. +configure_package_logger(__name__) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 8dce7d1e4ff..a0a39c790ba 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -153,7 +153,7 @@ def jobs(self) -> JobsList: else j for j in jobs_list ] - return JobsList(wrapped, jobs_list._root_email) + return JobsList(wrapped, jobs_list._root_email, jobs_list._has_do_role) def submit_python_job( self, @@ -266,13 +266,10 @@ def approve_job(self, job: JobInfo) -> None: if os.environ.get("PRE_SYNC", "true").lower() == "true": self._rds.sync() + # approve() refuses when the party's approval file is missing, so + # reaching the next line means there is a file to sync. job.approve() - file_name = enclave_approval_file_name(self.email) - approval_file = job.job_review_path / file_name - if not approval_file.exists(): - print( - "🟠 Approval file does not exist yet. Kindly wait until enclave sends it." - ) + approval_file = job.job_review_path / enclave_approval_file_name(self.email) relative_path = approval_file.relative_to(self._rds.syftbox_folder) self._rds.sync_engine.datasite_watcher_syncer.on_file_change( relative_path, process_now=True diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py index 287487a8e84..f086427cbfc 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py @@ -72,8 +72,11 @@ def approve(self) -> None: approval_file = self.job_review_path / file_name if not approval_file.exists(): raise PermissionError( - f"No approval file found for {self.current_user_email}. " - f"You may not be a designated party for this job." + f"No approval file for {self.current_user_email} on job " + f"'{self.name}'. The enclave writes one per designated party " + f"when it distributes the job, so either it has not distributed " + f"this job yet — run client.sync() and retry — or you are not a " + f"party to it." ) approval = PartyApprovalStatus.load_json(approval_file) if approval.status != JobStatus.PENDING: diff --git a/packages/syft-enclave/tests/test_enclave_job_info.py b/packages/syft-enclave/tests/test_enclave_job_info.py new file mode 100644 index 00000000000..2af7fd55813 --- /dev/null +++ b/packages/syft-enclave/tests/test_enclave_job_info.py @@ -0,0 +1,82 @@ +"""Unit tests for EnclaveJobInfo, the per-party approval gate. + +The gate lives here rather than in SyftEnclaveClient.approve_job, so these +build a job on a tmp_path SyftBox folder instead of a four-party enclave flow. +""" + +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from syft_enclaves.enclave_job_info import ( + EnclaveJobInfo, + PartyApprovalStatus, + enclave_approval_file_name, +) +from syft_job.client import JobClient +from syft_job.config import SyftJobConfig +from syft_job.job import JobInfo +from syft_job.job_storage import JobRef +from syft_job.models import JobState, JobStatus, JobSubmissionMetadata + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" + + +def _make_enclave_job(tmp_path: Path, job_name: str = "test_job") -> EnclaveJobInfo: + """An enclave job on the DO's datasite, with no approval file written yet.""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + ) + ref = JobRef( + datasite_email=DO_EMAIL, + ds_email=DS_EMAIL, + job_name=job_name, + protocol_version="1", + ) + job = JobInfo( + job_metadata=JobSubmissionMetadata( + name=job_name, + type="python", + submitted_by=DS_EMAIL, + datasite_email=DO_EMAIL, + submitted_at=datetime.now(timezone.utc), + ), + state=JobState(status=JobStatus.PENDING), + client=client, + current_user_email=DO_EMAIL, + ref=ref, + ) + return EnclaveJobInfo.from_job_info(job) + + +def test_approve_refuses_when_approval_file_missing(tmp_path: Path): + """No approval file means the enclave has not distributed the job yet. + + The message used to say the caller may not be a designated party, which is + the wrong cause for the common case and offers nothing to do about it. + """ + job = _make_enclave_job(tmp_path) + + with pytest.raises(PermissionError) as exc: + job.approve() + + message = str(exc.value) + assert DO_EMAIL in message + assert "test_job" in message + assert "client.sync()" in message + + +def test_approve_refuses_when_already_approved(tmp_path: Path): + """A second approval must not overwrite the first one's timestamp.""" + job = _make_enclave_job(tmp_path) + approval_file = job.job_review_path / enclave_approval_file_name(DO_EMAIL) + PartyApprovalStatus(party=DO_EMAIL).save_json(approval_file) + + job.approve() + assert PartyApprovalStatus.load_json(approval_file).status == JobStatus.APPROVED + + with pytest.raises(ValueError, match="Already in status: approved"): + job.approve() diff --git a/packages/syft-job/src/syft_job/__init__.py b/packages/syft-job/src/syft_job/__init__.py index 1341d77d48f..4bd49e496e3 100644 --- a/packages/syft-job/src/syft_job/__init__.py +++ b/packages/syft-job/src/syft_job/__init__.py @@ -1,5 +1,6 @@ # __version__ comes from the installed distribution metadata (see version.py). from .version import __version__ +from .logging_config import configure_package_logger from .client import BaseJobClient, JobClient, get_client from .config import SyftJobConfig @@ -32,3 +33,6 @@ # Migration registry "job_registry", ] + +# Last, so the imports stay at the top. Nothing here logs at import time. +configure_package_logger(__name__) diff --git a/packages/syft-job/src/syft_job/client.py b/packages/syft-job/src/syft_job/client.py index 1a7d7acaa9d..9f8066850f7 100644 --- a/packages/syft-job/src/syft_job/client.py +++ b/packages/syft-job/src/syft_job/client.py @@ -608,8 +608,18 @@ def jobs(self) -> JobsList: """ Get all jobs from all peer directories as an indexable list grouped by user. + Order is the same as the jobs table: jobs on your own datasite first, + then the other datasites alphabetically, newest-first within each one. + ``jobs[N]`` is the row labelled ``[N]``. Prefer an email and a name + (``jobs["do@org.com"]["analysis"]``); both parts stay the same as jobs + are added, and the email is what makes the name resolve to one job. + Returns a JobsList object that can be: - - Indexed: jobs[0], jobs[1], etc. + - Indexed by email, then name (preferred): jobs["do@org.com"]["analysis"] + - Indexed by datasite and submitter, for a name they share: + jobs["do@org.com"]["ds@org.com"]["analysis"] + - Indexed by name across every datasite: jobs["analysis"] + - Indexed by position: jobs[0], jobs[1] — matches the table Index column - Iterated: for job in jobs - Displayed: print(jobs) shows separate tables for each user - HTML display: in Jupyter, shows separate tables for each user with jobs @@ -622,25 +632,18 @@ def jobs(self) -> JobsList: current_jobs = self._get_all_jobs() - # Sort jobs by recent submissions first (newest first), then by user/status def job_sort_key(job): - # Parse submitted_at timestamp for sorting (most recent first) + # Root owner first, then peers, newest-first within each owner. + # This list is the authority for both jobs[N] and the table's [N]. try: if job.submitted_at: - from datetime import datetime as dt - - # Parse ISO format timestamp - ts = dt.fromisoformat(job.submitted_at.replace("Z", "+00:00")) - # Use negative timestamp for reverse chronological order (newest first) + ts = datetime.fromisoformat(job.submitted_at.replace("Z", "+00:00")) time_priority = -ts.timestamp() else: - # Jobs without submitted_at go to the end time_priority = float("inf") except Exception: - # Invalid timestamps go to the end time_priority = float("inf") - # Secondary sorting: user priority (root first), then user name, then status user_priority = ( 0 if job.datasite_owner_email == self.current_user_email else 1 ) @@ -656,9 +659,9 @@ def job_sort_key(job): status_priority = status_order.get(job.status, 7) return ( - time_priority, user_priority, job.datasite_owner_email, + time_priority, status_priority, job.name.lower(), ) diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 0eed2bbf7ad..f2c99a7d65a 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -1,6 +1,7 @@ from __future__ import annotations import shutil +import warnings from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, List, Optional @@ -120,7 +121,7 @@ def _list_output_files() -> List[Path]: for item in outputs_dir.iterdir() if item.name != PERMISSION_FILE_NAME ] - except Exception: + except OSError: return [] if status == JobStatus.FAILED: @@ -174,7 +175,7 @@ def files(self) -> List[Path]: ): continue all_files.append(f) - except Exception: + except OSError: pass return all_files @@ -213,8 +214,10 @@ def approve( if self.datasite_owner_email != self.current_user_email: raise PermissionError( - f"Only the admin user ({self.datasite_owner_email}) can approve jobs in their folder. " - f"Current job is in {self.datasite_owner_email}'s folder." + f"You are {self.current_user_email}, and job '{self.name}' is on " + f"{self.datasite_owner_email}'s datasite. Only they can approve " + f"it. If you meant one of your own, select your datasite " + f'first: jobs["{self.current_user_email}"][""].' ) self._state.status = JobStatus.APPROVED @@ -246,7 +249,8 @@ def reject(self, reason: Optional[str] = None) -> None: if self.datasite_owner_email != self.current_user_email: raise PermissionError( - f"Only the admin user ({self.datasite_owner_email}) can reject jobs." + f"You are {self.current_user_email}, and job '{self.name}' is on " + f"{self.datasite_owner_email}'s datasite. Only they can reject it." ) self._state.status = JobStatus.REJECTED @@ -423,6 +427,27 @@ def _repr_html_(self) -> str: return job_info_repr_html(self) +def _with_party(jobs: List[JobInfo], email: str) -> List[JobInfo]: + """The jobs ``email`` is a party to: on its datasite, or submitted by it.""" + return [j for j in jobs if email in (j.datasite_owner_email, j.submitted_by)] + + +def _with_name(jobs: List[JobInfo], name: str) -> List[JobInfo]: + """The jobs called ``name``.""" + return [j for j in jobs if j.name == name] + + +def _keep(jobs: List[JobInfo], key: str) -> List[JobInfo]: + """The jobs one subscript key keeps, read the way ``__getitem__`` reads it. + + The '@' tells the two kinds of key apart, so the usage hint can measure a + chain before offering it. A deprecated job name holding an '@' reads here + as an email and keeps nothing, where ``__getitem__`` falls back to the + name; the hint then offers a position, which is the safe direction to err. + """ + return _with_party(jobs, key) if "@" in key else _with_name(jobs, key) + + class JobsList: """A list-like container for JobInfo objects with nice display.""" @@ -431,17 +456,134 @@ def __init__(self, jobs: List[JobInfo], root_email: str, has_do_role: bool = Fal self._root_email = root_email self._has_do_role = has_do_role - def __getitem__(self, index: int | str) -> JobInfo: + def __getitem__(self, index: int | str) -> "JobInfo | JobsList": + """A job by position or name, or the jobs of one party by email. + + An email keeps the jobs it is a party to, on either side. That is + usually the other party — a data scientist names the data owner, a data + owner names the submitter — but naming yourself keeps your own. So + ``jobs["do@x.org"]["analysis"]`` reads as one job, and chaining both — + ``jobs["do@x.org"]["ds@y.org"]["analysis"]`` — pins the datasite and the + submitter, the pair a job name is unique under. A bare name searches + every datasite at once and raises when more than one job answers to it. + Job names cannot contain ``@``, so the two kinds of key never collide. + """ if isinstance(index, int): return self._jobs[index] elif isinstance(index, str): - for job in self._jobs: - if job.name == index: - return job - raise ValueError(f"Job with name '{index}' not found") + if "@" in index: + return self._by_email(index) + return self._by_name(index) else: raise TypeError(f"Invalid index type: {type(index)}") + def _by_email(self, email: str) -> "JobInfo | JobsList": + """The jobs this email is a party to: on its datasite, or submitted by it. + + One key covers both roles because which one narrows depends on who is + asking. A data scientist names the data owner's datasite; a data owner + names the submitter. Chaining the two pins the pair. + + A job submitted before names could not hold an '@' answers to no party, + and reading it is the one thing this key must not take away, so a key + that names no party falls back to the name. + """ + matches = _with_party(self._jobs, email) + if matches: + return JobsList(matches, self._root_email, self._has_do_role) + + if any(job.name == email for job in self._jobs): + warnings.warn( + f"Job name {email!r} holds an '@', which now marks a datasite or " + "submitter email. Such names are deprecated and the next version " + "will not resolve them. Rename the job.", + DeprecationWarning, + stacklevel=3, + ) + return self._by_name(email) + + datasites = ", ".join(sorted({job.datasite_owner_email for job in self._jobs})) + submitters = ", ".join(sorted({job.submitted_by for job in self._jobs})) + raise ValueError( + f"No jobs involving {email}. These jobs are on: {datasites}. " + f"They were submitted by: {submitters}." + ) + + def _by_name(self, name: str) -> JobInfo: + matches = _with_name(self._jobs, name) + if not matches: + raise ValueError(f"Job with name '{name}' not found") + if len(matches) > 1: + raise ValueError(self._ambiguous_name_message(name, matches)) + return matches[0] + + def _ambiguous_name_message(self, name: str, matches: List[JobInfo]) -> str: + """Why the name did not resolve, and the narrower subscript to use. + + A name is unique per datasite and submitter, not across the list, so + the remedy names whichever of the two separates these candidates. One + submitter holding a name twice across protocol layouts shares both, and + naming either would send the caller back to this same error, so the + position is the only thing left to offer. + """ + locations = ", ".join( + f"[{i}] on {job.datasite_owner_email} from {job.submitted_by}" + for i, job in enumerate(self._jobs) + if job.name == name + ) + if len({job.datasite_owner_email for job in matches}) > 1: + remedy = f'Select the datasite first: jobs[""]["{name}"].' + elif len({job.submitted_by for job in matches}) > 1: + remedy = f'Select the submitter first: jobs[""]["{name}"].' + else: + remedy = "One datasite and one submitter hold both, so no email " + remedy += "narrows them: select one by position." + return f"Multiple jobs are named '{name}': {locations}. {remedy}" + + def hint_accessor(self) -> str | None: + """The subscript chain the jobs table tells a data owner to type. + + Names a pending job on the DO's own datasite and gives the shortest + chain that reaches it and nothing else. The email in a chain is the + other party, so the submitter comes first; the datasite joins it only + when that submitter used the name on another datasite too. One + submitter holding a name twice across protocol layouts defeats every + chain, which is what the position is for. + + Returns None when no such job is there to name, because the hint's + ``approve()`` takes only a pending job on your own datasite: a client + with no data-owner role, a DO who owns none of these jobs, or a DO + whose own jobs have all been reviewed already. The hint also offers + ``accept_by_depositing_result()``, which an approved job would still + take; withholding both is the cost of never printing a command that + raises. + """ + if not self._has_do_role: + return None + owned = [ + j + for j in self._jobs + if j.datasite_owner_email == self._root_email and j.status == "pending" + ] + if not owned: + return None + pick = owned[0] + chains = ( + (pick.submitted_by, pick.name), + (self._root_email, pick.submitted_by, pick.name), + ) + for keys in chains: + if self._reaches_only(keys, pick): + return "".join(f'["{key}"]' for key in keys) + return f"[{self._jobs.index(pick)}]" + + def _reaches_only(self, keys: tuple[str, ...], job: JobInfo) -> bool: + """Whether subscripting by ``keys`` in turn reaches ``job`` and nothing else.""" + reached = self._jobs + for key in keys: + reached = _keep(reached, key) + return len(reached) == 1 and reached[0] is job + def __len__(self) -> int: return len(self._jobs) @@ -449,10 +591,10 @@ def __iter__(self): return iter(self._jobs) def __str__(self) -> str: - return jobs_list_str(self._jobs, self._root_email, self._has_do_role) + return jobs_list_str(self._jobs, self.hint_accessor()) def __repr__(self) -> str: return f"JobsList({len(self._jobs)} jobs)" def _repr_html_(self) -> str: - return jobs_list_repr_html(self._jobs, self._root_email, self._has_do_role) + return jobs_list_repr_html(self._jobs, self.hint_accessor()) diff --git a/packages/syft-job/src/syft_job/job_repr.py b/packages/syft-job/src/syft_job/job_repr.py index a12507aaab7..8c900b55281 100644 --- a/packages/syft-job/src/syft_job/job_repr.py +++ b/packages/syft-job/src/syft_job/job_repr.py @@ -731,18 +731,30 @@ def job_info_repr_html(job: "JobInfo") -> str: """ -def jobs_list_str( - jobs: List["JobInfo"], root_email: str, has_do_role: bool = False -) -> str: +def _owner_groups(jobs: List["JobInfo"]) -> List[tuple[str, List["JobInfo"]]]: + """Split jobs into consecutive owner sections without reordering. + + JobClient.jobs is the authority for order. Re-sorting here is what + made the table's [N] disagree with jobs[N]. A list that is not already + grouped by owner gives an owner more than one section; the row indexes + stay correct. + """ + groups: List[tuple[str, List["JobInfo"]]] = [] + for job in jobs: + owner = job.datasite_owner_email + if groups and groups[-1][0] == owner: + groups[-1][1].append(job) + else: + groups.append((owner, [job])) + return groups + + +def jobs_list_str(jobs: List["JobInfo"], hint_accessor: str | None = None) -> str: """Format jobs list as separate tables grouped by user.""" if not jobs: return "📭 No jobs found.\n" - jobs_by_user: dict[str, list["JobInfo"]] = {} - for job in jobs: - if job.datasite_owner_email not in jobs_by_user: - jobs_by_user[job.datasite_owner_email] = [] - jobs_by_user[job.datasite_owner_email].append(job) + owner_groups = _owner_groups(jobs) status_emojis = { "received": "📨", @@ -758,25 +770,12 @@ def jobs_list_str( lines.append("📊 Jobs Overview") lines.append("=" * 50) - total_jobs = 0 + total_jobs = len(jobs) global_status_counts: dict[str, int] = {} - def user_sort_key(item): - user_email, _user_jobs = item - if user_email == root_email: - return (0, user_email) - return (1, user_email) - - sorted_users = sorted(jobs_by_user.items(), key=user_sort_key) - job_index = 0 - for user_email, user_jobs in sorted_users: - if not user_jobs: - continue - - total_jobs += len(user_jobs) - + for user_email, user_jobs in owner_groups: lines.append("") lines.append(f"👤 {user_email}") lines.append("-" * 60) @@ -795,9 +794,7 @@ def user_sort_key(item): lines.append(header) lines.append("-" * len(header)) - sorted_jobs = user_jobs - - for job in sorted_jobs: + for job in user_jobs: emoji = status_emojis.get(job.status, "❓") status_display = f"{emoji} {job.status}" approval_display = job.approval_method or "—" @@ -824,7 +821,8 @@ def user_sort_key(item): lines.append("") lines.append("=" * 50) - lines.append(f"📈 Total: {total_jobs} jobs across {len(jobs_by_user)} users") + owner_count = len({job.datasite_owner_email for job in jobs}) + lines.append(f"📈 Total: {total_jobs} jobs across {owner_count} users") global_summary_parts = [] for status, count in global_status_counts.items(): @@ -834,18 +832,17 @@ def user_sort_key(item): if global_summary_parts: lines.append("📋 Global: " + " | ".join(global_summary_parts)) - if has_do_role: + if hint_accessor is not None: lines.append("") lines.append( - "💡 Use job_client.jobs[0].approve() to approve jobs or job_client.jobs[0].accept_by_depositing_result('file_or_folder') to complete jobs" + f"💡 Use job_client.jobs{hint_accessor}.approve() to approve jobs or " + f"job_client.jobs{hint_accessor}.accept_by_depositing_result('file_or_folder') to complete jobs" ) return "\n".join(lines) -def jobs_list_repr_html( - jobs: List["JobInfo"], root_email: str, has_do_role: bool = False -) -> str: +def jobs_list_repr_html(jobs: List["JobInfo"], hint_accessor: str | None = None) -> str: """HTML representation for Jupyter notebooks with enhanced visual appeal.""" if not jobs: return """ @@ -919,11 +916,7 @@ def jobs_list_repr_html( """ - jobs_by_user: dict[str, list["JobInfo"]] = {} - for job in jobs: - if job.datasite_owner_email not in jobs_by_user: - jobs_by_user[job.datasite_owner_email] = [] - jobs_by_user[job.datasite_owner_email].append(job) + owner_groups = _owner_groups(jobs) status_styles = { "received": { @@ -964,6 +957,7 @@ def jobs_list_repr_html( } total_jobs = len(jobs) + owner_count = len({job.datasite_owner_email for job in jobs}) global_status_counts: dict[str, int] = {} for job in jobs: global_status_counts[job.status] = global_status_counts.get(job.status, 0) + 1 @@ -1217,25 +1211,12 @@ def jobs_list_repr_html(

📊 Jobs Overview

-

Total: {total_jobs} jobs across {len(jobs_by_user)} users

+

Total: {total_jobs} jobs across {owner_count} users

""" - def user_sort_key(item): - user_email, _user_jobs = item - if user_email == root_email: - return (0, user_email) - return (1, user_email) - - sorted_users = sorted(jobs_by_user.items(), key=user_sort_key) - job_index = 0 - for user_email, user_jobs in sorted_users: - if not user_jobs: - continue - - sorted_user_jobs = user_jobs - + for user_email, user_jobs in owner_groups: user_status_counts: dict[str, int] = {} for job in user_jobs: user_status_counts[job.status] = user_status_counts.get(job.status, 0) + 1 @@ -1264,7 +1245,7 @@ def user_sort_key(item): """ - for i, job in enumerate(sorted_user_jobs): + for i, job in enumerate(user_jobs): style_info = status_styles.get(job.status, {"emoji": "❓"}) row_class = "syftjob-row-even" if i % 2 == 0 else "syftjob-row-odd" @@ -1314,10 +1295,10 @@ def user_sort_key(item): html += """
""" - if has_do_role: - html += """ + if hint_accessor is not None: + html += f"""
- 💡 Use jobs[0].approve() to approve jobs or jobs[0].accept_by_depositing_result('file_or_folder') to complete jobs + 💡 Use jobs{hint_accessor}.approve() to approve jobs or jobs{hint_accessor}.accept_by_depositing_result('file_or_folder') to complete jobs
""" html += """ diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 471eca5d811..21a56a63f59 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -2,6 +2,7 @@ import shutil import subprocess import time +import warnings from datetime import datetime, timezone from pathlib import Path from typing import List, Set @@ -515,32 +516,76 @@ def _get_job_info(self, ref: JobRef) -> JobInfo: ref=ref, ) + @staticmethod + def _resolve_skip_jobs( + approved_jobs: List[JobRef], + skip_jobs: list[tuple[str, str]] | None, + skip_job_names: list[str] | None, + ) -> Set[tuple[str, str]]: + """The (job_name, ds_email) pairs to skip, accepting the old name-only form. + + A bare name expands to every submitter who used it, which is what + ``skip_job_names`` did. Bare names in ``skip_jobs`` are read the same + way, so a caller who passed the old third positional argument keeps the + behaviour it had. + """ + pairs: Set[tuple[str, str]] = set() + names: List[str] = list(skip_job_names or []) + for entry in skip_jobs or []: + if isinstance(entry, str): + names.append(entry) + else: + pairs.add((entry[0], entry[1])) + + if names: + warnings.warn( + "Skipping jobs by name alone is deprecated: a name is unique " + "per datasite and submitter, so it also drops other " + "submitters' jobs of the same name. Pass " + "skip_jobs=[(job_name, ds_email), ...] instead.", + DeprecationWarning, + stacklevel=3, + ) + wanted = set(names) + pairs.update( + (ref.job_name, ref.ds_email) + for ref in approved_jobs + if ref.job_name in wanted + ) + return pairs + def process_approved_jobs( self, stream_output: bool = True, timeout: int | None = None, - skip_job_names: list[str] | None = None, + skip_jobs: list[tuple[str, str]] | None = None, share_outputs_with_submitter: bool = False, share_logs_with_submitter: bool = False, + skip_job_names: list[str] | None = None, ) -> None: """Process all jobs in approved status. Args: stream_output: If True (default), stream output in real-time. timeout: Timeout in seconds per job. Defaults to 300 (5 minutes). - skip_job_names: Optional list of job names to skip. + skip_jobs: Optional (job_name, ds_email) pairs to skip. A name alone + does not identify a job — it is unique per datasite and + submitter — so skipping by name drops every other job that + shares it. share_outputs_with_submitter: If True, grant read access on outputs to submitter. share_logs_with_submitter: If True, grant read access on logs to submitter. + skip_job_names: Deprecated. The name-only form of ``skip_jobs``. """ approved_jobs = self._get_jobs_in_approved() if not approved_jobs: return - # Filter out jobs to skip - if skip_job_names: - skip_set = set(skip_job_names) - approved_jobs = [j for j in approved_jobs if j.job_name not in skip_set] + skip_set = self._resolve_skip_jobs(approved_jobs, skip_jobs, skip_job_names) + if skip_set: + approved_jobs = [ + j for j in approved_jobs if (j.job_name, j.ds_email) not in skip_set + ] if not approved_jobs: return diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index c0b5c403f96..f8d116ad56e 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -141,6 +141,10 @@ def validate_job_name(job_name: str) -> None: raise ValueError( f"Job name {job_name!r} is reserved for protocol version directories" ) + if "@" in job_name: + # An '@' marks a datasite email in JobsList.__getitem__, which is + # how jobs["do@x.org"]["analysis"] tells the two keys apart. + raise ValueError(f"Job name {job_name!r} cannot contain '@'") # -- scanning (union over all protocol layouts) ------------------------------- def iter_submission_refs(self, datasite_email: str) -> Iterator[JobRef]: diff --git a/packages/syft-job/src/syft_job/logging_config.py b/packages/syft-job/src/syft_job/logging_config.py new file mode 100644 index 00000000000..6096c3a9e92 --- /dev/null +++ b/packages/syft-job/src/syft_job/logging_config.py @@ -0,0 +1,31 @@ +"""Shared logger setup for syft-job and the packages built on it. + +``syft`` configures the ``syft`` logger only. ``syft_job``, ``syft_rds``, +``syft_enclaves`` and ``syft_bg`` are sibling top-level namespaces, so they +inherit the root logger, which has no handler. INFO records are then dropped, +and WARNING and above fall back to ``logging.lastResort`` — bare text on +stderr, with no way to raise or lower the level. Each package calls +``configure_package_logger`` once, at the end of its ``__init__``. + +The helper lives here, not in ``syft``, because syft-job does not depend on +syft, while the other three packages depend on syft-job. +""" + +import logging + + +def configure_package_logger(name: str, level: int = logging.INFO) -> logging.Logger: + """Give the ``name`` logger a handler and a level, but only if it has none. + + A caller who sets up their own logging keeps full control. Records still + propagate to the root logger, so pytest's caplog and any root handler still + see them; in default Python the root logger has no handler, so nothing is + printed twice. + """ + logger = logging.getLogger(name) + if not logger.handlers: + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter("[%(levelname)s] %(message)s")) + logger.addHandler(handler) + logger.setLevel(level) + return logger diff --git a/packages/syft-job/tests/test_job_flow.py b/packages/syft-job/tests/test_job_flow.py index d5022e76417..2aba9c2d9cc 100644 --- a/packages/syft-job/tests/test_job_flow.py +++ b/packages/syft-job/tests/test_job_flow.py @@ -3,15 +3,17 @@ import time from pathlib import Path +import pytest from syft_job.client import JobClient from syft_job.config import SyftJobConfig from syft_job.job_runner import SyftJobRunner -from syft_perms import SyftPermContext from syft_job.models import JobState - +from syft_perms import SyftPermContext DO_EMAIL = "do@test.org" DS_EMAIL = "ds@test.org" +PEER_EMAIL = "peer@test.org" +DS2_EMAIL = "ds2@test.org" MAIN_PY = """\ import os @@ -252,3 +254,172 @@ def test_timeout_does_not_hang_runner(tmp_path: Path): # 3s job timeout + venv setup + tree-kill cleanup should fit well under 60s. assert elapsed < 60, f"process_approved_jobs took {elapsed:.1f}s — likely hung" assert do_client.jobs[0].status == "failed" + + +def test_skipping_one_job_spares_its_same_named_sibling(tmp_path: Path): + """A skip must name the submitter as well as the job. + + Two data scientists can submit the same job name to one data owner. Given + only the name, the runner dropped both — so a job its peer was compatible + with never ran, and nothing said so. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + do_client = JobClient(config=do_config) + do_runner = SyftJobRunner(config=do_config) + + for ds_email in (DS_EMAIL, DS2_EMAIL): + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=ds_email) + ).submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="shared.job" + ) + + for job in do_client.jobs: + job.approve() + + do_runner.process_approved_jobs( + stream_output=False, timeout=60, skip_jobs=[("shared.job", DS_EMAIL)] + ) + + final = {job.submitted_by: job.status for job in do_client.jobs} + assert final[DS_EMAIL] == "approved", "the skipped job must not run" + assert final[DS2_EMAIL] == "done", "its same-named sibling must still run" + + +def test_runner_ignores_approved_jobs_on_another_datasite(tmp_path: Path): + """`jobs` spans every datasite in the folder; the runner runs only its own. + + A peer's approved job is visible but was never a candidate, which is why the + skip report in syft-rds filters on the datasite owner before naming a job as + one that did not run. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + peer_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=PEER_EMAIL) + do_client = JobClient(config=do_config) + peer_client = JobClient(config=peer_config) + do_runner = SyftJobRunner(config=do_config) + + ds_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="mine.job" + ) + ds_client.submit_python_job( + user=PEER_EMAIL, code_path=str(code_file), job_name="theirs.job" + ) + + do_client.jobs["mine.job"].approve() + peer_client.jobs["theirs.job"].approve() + + # The DO sees both, including the peer's approved job. + visible = {job.name: job.status for job in do_client.jobs} + assert visible == {"mine.job": "approved", "theirs.job": "approved"} + + do_runner.process_approved_jobs(stream_output=False, timeout=60) + + after = {job.name: job.status for job in do_client.jobs} + assert after["mine.job"] == "done" + assert after["theirs.job"] == "approved", "a peer's job is not this runner's to run" + + +def test_deprecated_skip_job_names_still_skips_by_name(tmp_path: Path): + """The old name-only argument keeps its old behaviour, and says it is old. + + It drops every submitter's job of that name — which is the bug — so it warns + rather than silently changing what an existing caller gets. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + do_client = JobClient(config=do_config) + do_runner = SyftJobRunner(config=do_config) + + for ds_email in (DS_EMAIL, DS2_EMAIL): + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=ds_email) + ).submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="shared.job" + ) + for job in do_client.jobs: + job.approve() + + with pytest.warns(DeprecationWarning, match="Skipping jobs by name alone"): + do_runner.process_approved_jobs( + stream_output=False, timeout=60, skip_job_names=["shared.job"] + ) + + assert [job.status for job in do_client.jobs] == ["approved", "approved"] + + +def test_approval_error_names_both_parties(tmp_path: Path): + """The old message called the peer "the admin user" and never named you.""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + ) + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ).submit_python_job( + user=PEER_EMAIL, code_path=str(code_file), job_name="peer-owned.job" + ) + # The owner scans it into pending, so approve() reaches the ownership check. + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=PEER_EMAIL) + ).scan_inbox() + + job = do_client.jobs["peer-owned.job"] + assert job.status == "pending" + with pytest.raises(PermissionError) as exc: + job.approve() + + message = str(exc.value) + assert DO_EMAIL in message, "must say who you are" + assert PEER_EMAIL in message, "must say whose datasite it is" + assert "peer-owned.job" in message + assert "admin user" not in message + # The remedy must not send them back to the same job: a bare name searches + # every datasite, and this job is the only one that answers to this one. + assert f'jobs["{DO_EMAIL}"]' in message, "must point at your own datasite" + + +def test_job_files_do_not_hide_non_filesystem_error(tmp_path: Path, monkeypatch): + """Only the filesystem may truncate the file list; other errors propagate.""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + ) + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ).submit_python_job(user=DO_EMAIL, code_path=str(code_file), job_name="files.job") + + job = do_client.jobs["files.job"] + assert job.files, "sanity: the job has files" + + def boom(*args, **kwargs): + raise RuntimeError("not a filesystem problem") + + monkeypatch.setattr(Path, "rglob", boom) + with pytest.raises(RuntimeError, match="not a filesystem problem"): + job.files diff --git a/packages/syft-job/tests/test_jobs_addressing.py b/packages/syft-job/tests/test_jobs_addressing.py new file mode 100644 index 00000000000..2f6be963f04 --- /dev/null +++ b/packages/syft-job/tests/test_jobs_addressing.py @@ -0,0 +1,338 @@ +"""How a job is addressed: the subscript chain, and the hint that teaches it. + +These drive JobClient over a tmp_path SyftBox folder, like the lifecycle tests +next door, but assert on what jobs[...] returns and what the table's hint says +rather than on a job running. +""" + +import re +from pathlib import Path + +import pytest +from syft_job.client import JobClient +from syft_job.config import SyftJobConfig + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" +PEER_EMAIL = "peer@test.org" +DS2_EMAIL = "ds2@test.org" + +MAIN_PY = """\ +import os + +print("hello from job") +os.makedirs("outputs", exist_ok=True) +with open("outputs/result.txt", "w") as f: + f.write("done") +""" + + +def test_hint_names_submitter_and_job(tmp_path: Path): + """The DO hint must point at jobs["submitter"]["name"], not jobs[0]. + + Positional indexing is not safe to recommend: positions shift as jobs are + added. A bare name is not safe either — it searches every datasite, so a + peer's job of the same name can answer instead. The email in a chain is the + other party, which for a data owner is the submitter. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + # has_do_role gates the hint — it is only shown to a data owner. + do_config = SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + ds_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ds_client = JobClient(config=ds_config) + do_client = JobClient(config=do_config) + + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="analysis.job" + ) + + jobs = do_client.jobs + text = str(jobs) + html = jobs._repr_html_() + + for rendering in (text, html): + assert f'jobs["{DS_EMAIL}"]["analysis.job"].approve()' in rendering + assert "jobs[0]" not in rendering + assert DO_EMAIL not in rendering.split("💡")[1], "the DO's own email is noise" + + # The hint names a job the DO can actually approve, as written. + assert do_client.jobs[DS_EMAIL]["analysis.job"].status == "pending" + + +def _index_name_pairs_from_text(text: str) -> list[tuple[int, str]]: + return [(int(i), name) for i, name in re.findall(r"\[(\d+)\s*\]\s+(\S+)", text)] + + +def _index_name_pairs_from_html(html: str) -> list[tuple[int, str]]: + return [ + (int(i), name.strip()) + for i, name in re.findall( + r'class="syftjob-index">\[(\d+)\].*?' + r'class="syftjob-td syftjob-job-name">\s*([^<]+)', + html, + flags=re.DOTALL, + ) + ] + + +def test_jobs_table_index_matches_getitem(tmp_path: Path): + """The [N] printed in the table must be the N that jobs[N] returns. + + The table groups by datasite owner (root first, then peers). __getitem__ + used to subscript a newest-first list, so with two owners the row labelled + [0] was not jobs[0]. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + ds_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ds_client = JobClient(config=ds_config) + do_client = JobClient(config=do_config) + + # Older job on the root datasite, then a newer one on a peer datasite. + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="older-root-job" + ) + ds_client.submit_python_job( + user=PEER_EMAIL, code_path=str(code_file), job_name="newest-peer-job" + ) + + jobs = do_client.jobs + assert [job.name for job in jobs] == ["older-root-job", "newest-peer-job"] + + text_pairs = _index_name_pairs_from_text(str(jobs)) + html_pairs = _index_name_pairs_from_html(jobs._repr_html_()) + assert text_pairs == [(0, "older-root-job"), (1, "newest-peer-job")] + assert html_pairs == [(0, "older-root-job"), (1, "newest-peer-job")] + + for index, name in text_pairs + html_pairs: + assert jobs[index].name == name + + +def test_hint_names_submitter_for_shared_name(tmp_path: Path): + """A name two submitters share needs the submitter to reach one job. + + Job names are unique per datasite and submitter, so two data scientists can + submit "analysis" to the same data owner. The hint adds the submitter rather + than giving up on the name. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + do_client = JobClient(config=do_config) + ds1_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + ds2_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS2_EMAIL) + ) + + # "solo" first, so the ambiguous name is the newest and would otherwise win. + ds1_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="solo" + ) + ds1_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="analysis" + ) + ds2_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="analysis" + ) + + jobs = do_client.jobs + for rendering in (str(jobs), jobs._repr_html_()): + assert f'jobs["{DS2_EMAIL}"]["analysis"].approve()' in rendering + + # The chain the hint gives reaches exactly one job, and the datasite alone + # does not, which is why it names the submitter. + job = jobs[DS2_EMAIL]["analysis"] + assert (job.submitted_by, job.status) == (DS2_EMAIL, "pending") + with pytest.raises(ValueError, match="Multiple jobs are named 'analysis'"): + jobs[DO_EMAIL]["analysis"] + + +def test_hint_names_every_submitter_of_shared_name(tmp_path: Path): + """Every job in the table shares its name, so every chain needs a submitter. + + The datasite alone reaches neither, and the hint still has to name something + the data owner can run. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + do_client = JobClient(config=do_config) + for ds_email in (DS_EMAIL, DS2_EMAIL): + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=ds_email) + ).submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="analysis" + ) + + jobs = do_client.jobs + for rendering in (str(jobs), jobs._repr_html_()): + assert f'jobs["{DS2_EMAIL}"]["analysis"].approve()' in rendering + assert "jobs[0]" not in rendering + + for ds_email in (DS_EMAIL, DS2_EMAIL): + assert jobs[ds_email]["analysis"].submitted_by == ds_email + + +def test_no_hint_when_do_owns_no_jobs(tmp_path: Path): + """Every subscript would name a job on another datasite, which approve() refuses. + + A hint is worse than no hint when the only command it can give raises + PermissionError. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_config = SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + do_client = JobClient(config=do_config) + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ).submit_python_job( + user=PEER_EMAIL, code_path=str(code_file), job_name="peer-owned.job" + ) + + jobs = do_client.jobs + assert [job.datasite_owner_email for job in jobs] == [PEER_EMAIL] + for rendering in (str(jobs), jobs._repr_html_()): + assert "peer-owned.job" in rendering + assert "💡" not in rendering + + +def test_email_key_resolves_name_shared_by_two_datasites(tmp_path: Path): + """One submitter can send the same job name to two data owners. + + The bare name has no way to choose between them and raises. Selecting the + datasite first leaves one job, which is the whole point of the two-step + subscript. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + ds_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + for owner in (DO_EMAIL, PEER_EMAIL): + ds_client.submit_python_job( + user=owner, code_path=str(code_file), job_name="analysis.job" + ) + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=owner) + ).scan_inbox() + + jobs = ds_client.jobs + with pytest.raises(ValueError, match="Multiple jobs are named 'analysis.job'"): + jobs["analysis.job"] + + for owner in (DO_EMAIL, PEER_EMAIL): + job = jobs[owner]["analysis.job"] + assert job.datasite_owner_email == owner + + +def test_email_key_names_emails_it_has(tmp_path: Path): + """An email with no jobs on it is a typo the message has to help with.""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + ds_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="analysis.job" + ) + + with pytest.raises(ValueError) as exc: + ds_client.jobs["nobody@test.org"] + + message = str(exc.value) + assert "nobody@test.org" in message + assert DO_EMAIL in message, "must name the datasites that do have jobs" + assert DS_EMAIL in message, "must name the submitters too" + + +def test_hint_adds_datasite_when_submitter_used_name_twice( + tmp_path: Path, +): + """One submitter can send the same name to the DO and to a peer. + + The submitter alone then reaches both, so this is the one case where the + DO's own datasite has to join the chain, and the only thing that earns the + third key. + """ + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + do_client = JobClient( + config=SyftJobConfig( + syftbox_folder=syftbox, current_user_email=DO_EMAIL, has_do_role=True + ) + ) + ds_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + for owner in (DO_EMAIL, PEER_EMAIL): + ds_client.submit_python_job( + user=owner, code_path=str(code_file), job_name="analysis.job" + ) + JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=owner) + ).scan_inbox() + + jobs = do_client.jobs + assert {job.datasite_owner_email for job in jobs} == {DO_EMAIL, PEER_EMAIL} + chain = f'jobs["{DO_EMAIL}"]["{DS_EMAIL}"]["analysis.job"]' + for rendering in (str(jobs), jobs._repr_html_()): + assert f"{chain}.approve()" in rendering + + # The submitter alone reaches both datasites, so the chain needs both keys. + with pytest.raises(ValueError, match="Multiple jobs are named 'analysis.job'"): + jobs[DS_EMAIL]["analysis.job"] + assert jobs[DO_EMAIL][DS_EMAIL]["analysis.job"].datasite_owner_email == DO_EMAIL + + +def test_job_name_cannot_contain_at_sign(tmp_path: Path): + """An '@' is how a subscript tells a datasite email from a job name.""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + code_file = tmp_path / "main.py" + code_file.write_text(MAIN_PY) + + ds_client = JobClient( + config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ) + with pytest.raises(ValueError, match="cannot contain '@'"): + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="ds@test.org" + ) diff --git a/packages/syft-rds/src/syft_rds/__init__.py b/packages/syft-rds/src/syft_rds/__init__.py index d4826ca6b8a..c86d6b6c62d 100644 --- a/packages/syft-rds/src/syft_rds/__init__.py +++ b/packages/syft-rds/src/syft_rds/__init__.py @@ -1,5 +1,7 @@ """syft-rds: Remote Data Science product composed on top of syft.""" +from syft_job.logging_config import configure_package_logger + from syft_rds.client import SyftRDSClient from syft_rds.config import SyftRDSClientConfig from syft_rds.job_auto_approval import auto_approve_and_run_jobs, job_matches_criteria @@ -25,3 +27,6 @@ "Environment", "check_env", ] + +# Last, so the imports stay at the top. Nothing here logs at import time. +configure_package_logger(__name__) diff --git a/packages/syft-rds/src/syft_rds/client.py b/packages/syft-rds/src/syft_rds/client.py index 3af8076ccd6..d220aebf47d 100644 --- a/packages/syft-rds/src/syft_rds/client.py +++ b/packages/syft-rds/src/syft_rds/client.py @@ -32,6 +32,14 @@ logger = logging.getLogger(__name__) +def _print_skipped_jobs(skipped: list[tuple[str, str, str]]) -> None: + """Report the approved jobs that did not run, and why.""" + print(f"\n⏭️ {len(skipped)} approved job(s) did not run:") + for job_name, peer_email, reason in skipped: + print(f" • {job_name} (submitted by {peer_email}): {reason}") + print(" Pass ignore_peer_version=True to run them anyway.") + + class SyftRDSClient(BaseModel): # Holds live service objects (sync engine + RDS-owned managers), not # serializable data, so arbitrary types are allowed. @@ -371,11 +379,18 @@ def process_approved_jobs( if self.job_runner is None: raise ValueError("Job runner is not configured for this client") - skip_job_names = [] + skipped: list[tuple[str, str, str]] = [] if not force_execution: + # Only jobs on this client's own datasite: job_client.jobs spans + # every datasite in the folder, but the runner never runs a job on + # a peer's. Reporting those as skipped would offer a remedy that + # cannot help. approved_jobs = [ - job for job in self.job_client.jobs if job.status == "approved" + job + for job in self.job_client.jobs + if job.status == "approved" + and job.datasite_owner_email == self.job_client.current_user_email ] for job in approved_jobs: result = self.sync_engine.peer_manager.get_peer_compatibility_status( @@ -385,16 +400,26 @@ def process_approved_jobs( ) result.maybe_warn() if result.should_skip: - skip_job_names.append(job.name) + skipped.append( + ( + job.name, + job.submitted_by, + result.explanation_skip or "peer version is not compatible", + ) + ) self.job_runner.process_approved_jobs( stream_output=stream_output, timeout=timeout, - skip_job_names=skip_job_names if skip_job_names else None, + skip_jobs=[(name, peer) for name, peer, _ in skipped] or None, share_outputs_with_submitter=share_outputs_with_submitter, share_logs_with_submitter=share_logs_with_submitter, ) + # maybe_warn() above logs the peer, never the job. + if skipped: + _print_skipped_jobs(skipped) + if self._pre_sync_enabled: self.sync_engine.sync() diff --git a/packages/syft-rds/tests/test_datasets_jobs_repr.py b/packages/syft-rds/tests/test_datasets_jobs_repr.py index 8d63b88cfca..18f5c0e7e8b 100644 --- a/packages/syft-rds/tests/test_datasets_jobs_repr.py +++ b/packages/syft-rds/tests/test_datasets_jobs_repr.py @@ -1,5 +1,7 @@ """Tests for SyftDatasetManager and JobsList repr and indexing.""" +import warnings + import pytest from syft_rds import SyftRDSClient @@ -134,7 +136,12 @@ def test_dataset_repr_html_mentions_mock_files(): # --- JobsList tests --- -def _make_job_info(name: str, status: str = "pending") -> JobInfo: +def _make_job_info( + name: str, + status: str = "pending", + ds_email: str = "ds@test.com", + owner_email: str = "test@test.com", +) -> JobInfo: """Create a minimal JobInfo for testing.""" from datetime import datetime, timezone from pathlib import Path @@ -151,15 +158,15 @@ def _make_job_info(name: str, status: str = "pending") -> JobInfo: submission_config = JobSubmissionMetadata( name=name, type="python", - submitted_by="ds@test.com", - datasite_email="ds@test.com", + submitted_by=ds_email, + datasite_email=ds_email, submitted_at=datetime.now(timezone.utc), ) state = JobState(status=JobStatus(status)) # Identity (owner, submitter, name) comes from the path-derived ref. ref = JobRef( - datasite_email="test@test.com", - ds_email="ds@test.com", + datasite_email=owner_email, + ds_email=ds_email, job_name=name, protocol_version="1", ) @@ -198,6 +205,241 @@ def test_jobs_list_getitem_str_not_found(): jobs["nonexistent"] +def test_jobs_list_getitem_str_ambiguous(): + """Two submitters can use one job name; the lookup must not guess between them. + + Names are unique per datasite and submitter, so the same name can appear + more than once in one list. Returning the first match approves the wrong + job and says nothing. + """ + jobs = JobsList( + [ + _make_job_info("analysis", ds_email="ds1@test.com"), + _make_job_info("analysis", ds_email="ds2@test.com"), + ], + root_email="test@test.com", + ) + with pytest.raises(ValueError, match="Multiple jobs are named 'analysis'") as exc: + jobs["analysis"] + + message = str(exc.value) + assert "[0] on test@test.com from ds1@test.com" in message + assert "[1] on test@test.com from ds2@test.com" in message + # One datasite holds both, so the submitter is what separates them. + assert 'jobs[""]["analysis"]' in message + + +def test_jobs_list_getitem_str_ambiguous_names_the_datasite(): + """One submitter can send the same job name to two data owners. + + The submitter is then identical on every row, so the message has to name + the datasite as well. This is the DS-side call the README recommends. + """ + jobs = JobsList( + [ + _make_job_info("analysis", owner_email="do1@test.com"), + _make_job_info("analysis", owner_email="do2@test.com"), + ], + root_email="ds@test.com", + ) + with pytest.raises(ValueError) as exc: + jobs["analysis"] + + message = str(exc.value) + assert "[0] on do1@test.com" in message + assert "[1] on do2@test.com" in message + assert 'jobs[""]["analysis"]' in message, ( + "the datasite narrows this" + ) + + +def test_jobs_list_getitem_email_selects_datasite(): + """An email key narrows to one datasite, where the name is unique.""" + jobs = JobsList( + [ + _make_job_info("analysis", owner_email="do1@test.com"), + _make_job_info("analysis", owner_email="do2@test.com"), + ], + root_email="ds@test.com", + ) + on_do1 = jobs["do1@test.com"] + assert isinstance(on_do1, JobsList) + assert len(on_do1) == 1 + assert on_do1["analysis"].datasite_owner_email == "do1@test.com" + + +def test_jobs_list_getitem_email_not_found(): + jobs = JobsList( + [_make_job_info("analysis", owner_email="do1@test.com")], + root_email="ds@test.com", + ) + with pytest.raises(ValueError, match="No jobs involving nobody@test.com"): + jobs["nobody@test.com"] + + +def test_jobs_list_getitem_submitter_narrows_one_datasite(): + """Two submitters to one datasite share a name the datasite cannot separate. + + Chaining the submitter is what reaches one job, and without it the lookup + must keep raising rather than guess. + """ + jobs = JobsList( + [ + _make_job_info("analysis", ds_email="ds1@test.com"), + _make_job_info("analysis", ds_email="ds2@test.com"), + ], + root_email="test@test.com", + ) + with pytest.raises(ValueError, match="Multiple jobs are named 'analysis'"): + jobs["test@test.com"]["analysis"] + + job = jobs["test@test.com"]["ds2@test.com"]["analysis"] + assert job.submitted_by == "ds2@test.com" + + +def test_jobs_list_getitem_legacy_name_with_at_warns(): + """A job named before the '@' ban must stay readable, and say it is on notice. + + Nothing warned the submitter at the time, so the name is already on disk. + The email key answers first; only a key no party answers to falls back. + """ + jobs = JobsList( + [ + _make_job_info( + "report@2026-09", owner_email="do@test.com", ds_email="ds@test.com" + ) + ], + root_email="do@test.com", + ) + with pytest.warns(DeprecationWarning, match="deprecated"): + job = jobs["report@2026-09"] + assert job.name == "report@2026-09" + + # A real party still wins the key, and warns about nothing. + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert len(jobs["do@test.com"]) == 1 + + +def test_hint_prefers_pending_job_over_unique_name(): + """The hint's approve() only accepts a pending job, so status wins. + + A finished job with a name no one shares once won the pick, and handed the + data owner a command that raises on status. + """ + jobs = JobsList( + [ + _make_job_info("solo", ds_email="ds1@test.com", status="done"), + _make_job_info("analysis", ds_email="ds1@test.com", status="pending"), + _make_job_info("analysis", ds_email="ds2@test.com", status="pending"), + ], + root_email="test@test.com", + has_do_role=True, + ) + assert jobs.hint_accessor() == '["ds1@test.com"]["analysis"]' + + +def test_hint_chain_reaches_job_it_names(): + """Whatever chain the hint gives, subscripting by it must return that job. + + The hint and the lookup read a chain through the same evaluator, so this + holds for the two-key and three-key forms alike. + """ + jobs = JobsList( + [ + _make_job_info( + "analysis", owner_email="do@test.com", ds_email="ds@test.com" + ), + _make_job_info( + "analysis", owner_email="peer@test.com", ds_email="ds@test.com" + ), + ], + root_email="do@test.com", + has_do_role=True, + ) + # One submitter, two datasites: the chain needs all three keys. + assert jobs.hint_accessor() == '["do@test.com"]["ds@test.com"]["analysis"]' + + reached = jobs["do@test.com"]["ds@test.com"]["analysis"] + assert reached.datasite_owner_email == "do@test.com" + + +def test_hint_gives_position_when_no_chain_resolves(): + """One submitter can hold a name twice across protocol layouts. + + Datasite and submitter are then identical on both rows, so no chain reaches + one job and the position is all that is left. + """ + jobs = JobsList( + [ + _make_job_info("analysis", ds_email="ds1@test.com"), + _make_job_info("analysis", ds_email="ds1@test.com"), + ], + root_email="test@test.com", + has_do_role=True, + ) + assert jobs.hint_accessor() == "[0]" + + +def test_ambiguous_name_offers_position_when_no_email_narrows(): + """One submitter can hold a name twice across protocol layouts. + + Datasite and submitter are identical on both rows, so naming either would + return the caller to this same error. Only the position separates them. + """ + jobs = JobsList( + [ + _make_job_info("analysis", ds_email="ds1@test.com"), + _make_job_info("analysis", ds_email="ds1@test.com"), + ], + root_email="test@test.com", + ) + with pytest.raises(ValueError) as exc: + jobs["analysis"] + + message = str(exc.value) + assert "by position" in message + assert "Select the submitter" not in message, "the submitter does not narrow this" + assert "Select the datasite" not in message + + +def test_no_hint_when_no_owned_job_is_pending(): + """The hint's approve() takes a pending job, so a reviewed list gets no hint. + + Every job here is finished, and naming one would print a command that + raises on status rather than on ownership. + """ + jobs = JobsList( + [ + _make_job_info("analysis", status="done"), + _make_job_info("failed-one", status="failed"), + _make_job_info("approved-one", status="approved"), + ], + root_email="test@test.com", + has_do_role=True, + ) + assert jobs.hint_accessor() is None + + +def test_jobs_list_repr_counts_distinct_owners(): + """The owner total counts owners, not table sections. + + The renderers group consecutive rows to keep the table order equal to the + list order. A list that is not grouped by owner gives an owner more than + one section, which must not inflate the total. + """ + jobs = JobsList( + [ + _make_job_info("job-a", owner_email="do1@test.com"), + _make_job_info("job-b", owner_email="do2@test.com"), + _make_job_info("job-c", owner_email="do1@test.com"), + ], + root_email="do1@test.com", + ) + assert "3 jobs across 2 users" in str(jobs) + assert "3 jobs across 2 users" in jobs._repr_html_() + + def test_jobs_list_getitem_invalid_type(): jobs = JobsList( [_make_job_info("job-a")], diff --git a/packages/syft-rds/tests/test_version_negotiation.py b/packages/syft-rds/tests/test_version_negotiation.py index aaf422a92b5..56da05bd1b5 100644 --- a/packages/syft-rds/tests/test_version_negotiation.py +++ b/packages/syft-rds/tests/test_version_negotiation.py @@ -1,9 +1,12 @@ """Version gating on the RDS job submission and execution paths.""" +from unittest.mock import patch + import pytest from syft.sync.version.exceptions import VersionUnknownError -from syft.sync.version.version_info import VersionInfo +from syft.sync.version.peer_manager import PeerCompatibilityResult, PeerManager +from syft.sync.version.version_info import CompatibilityStatus, VersionInfo from syft_rds import SyftRDSClient @@ -113,9 +116,9 @@ def test_job_execution_forced_with_incompatible_version(self): executed_jobs = [] def mock_process_approved_jobs( - stream_output=True, timeout=None, skip_job_names=None, **kwargs + stream_output=True, timeout=None, skip_jobs=None, **kwargs ): - executed_jobs.append(skip_job_names) + executed_jobs.append(skip_jobs) do.job_runner.process_approved_jobs = mock_process_approved_jobs @@ -123,3 +126,44 @@ def mock_process_approved_jobs( assert len(executed_jobs) == 1 assert executed_jobs[0] is None # No jobs skipped when force=True + + +class TestSkippedJobsAreReported: + """process_approved_jobs must say which approved jobs it did not run. + + maybe_warn() logs the peer, not the job, so on its own it leaves the data + owner with a job stuck at 'approved' and nothing naming it. + """ + + def test_skipped_job_name_and_reason_reach_stdout(self, tmp_path, capfd): + ds, do = SyftRDSClient.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + sync_automatically=False, + ) + + code_path = tmp_path / "skipped.py" + code_path.write_text('print("hello")') + ds.submit_python_job( + user=do.email, code_path=str(code_path), job_name="skipped.job" + ) + do.sync() + do.jobs["skipped.job"].approve() + + skip = PeerCompatibilityResult( + peer_email=ds.email, + status=CompatibilityStatus.INCOMPATIBLE, + should_skip=True, + explanation_skip=f"Skipping peer {ds.email}: incompatible version.", + ) + capfd.readouterr() + with patch.object( + PeerManager, "get_peer_compatibility_status", return_value=skip + ): + do.process_approved_jobs() + out, _ = capfd.readouterr() + + assert "skipped.job" in out + assert ds.email in out + assert "incompatible version" in out + assert "ignore_peer_version=True" in out + assert do.jobs["skipped.job"].status == "approved" diff --git a/syft/__init__.py b/syft/__init__.py index 04033450b27..5a58db3189d 100644 --- a/syft/__init__.py +++ b/syft/__init__.py @@ -20,6 +20,7 @@ from syft.version import SYFT_VERSION as __version__ # noqa: F401, E402 from syft.sync.login import login_do, login_ds, login # noqa: F401, E402 +from syft.sync.peers.exceptions import SyftPeerNotReadyError # noqa: F401, E402 from syft.utils import ( # noqa: F401, E402 resolve_path, resolve_dataset_file_path, diff --git a/syft/sync/peers/exceptions.py b/syft/sync/peers/exceptions.py new file mode 100644 index 00000000000..db2aa7855c0 --- /dev/null +++ b/syft/sync/peers/exceptions.py @@ -0,0 +1,18 @@ +"""Peer-related exceptions for syft.""" + + +class SyftPeerNotReadyError(ValueError): + """A peer is known but not yet usable for the operation. + + Carries the cause and the remedy separately so callers — a readiness + helper, or a diagnostic — can show either one on its own. + + Subclasses ValueError because these conditions were reported as a bare + ValueError before, and callers catch that. + """ + + def __init__(self, peer_email: str, cause: str, remedy: str): + self.peer_email = peer_email + self.cause = cause + self.remedy = remedy + super().__init__(f"{cause} {remedy}") diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index eb8ef855497..70df8a2384f 100644 --- a/syft/sync/peers/peer_store.py +++ b/syft/sync/peers/peer_store.py @@ -11,7 +11,8 @@ import syft_crypto_python as syc from pydantic import BaseModel, PrivateAttr -from syft.sync.peers.peer import Peer +from syft.sync.peers.exceptions import SyftPeerNotReadyError +from syft.sync.peers.peer import Peer, PeerState # Encryption key bundles persist inside the participant's own SyftBox datasite # folder, under private/ (which is never synced to Drive). This scopes keys per @@ -128,21 +129,65 @@ def set_peers(self, peers: List[Peer]) -> None: def _ensure_private_keys(self) -> syc.SyftPrivateKeys: if self._private_keys is None: - raise ValueError("No private keys — call generate_keys() first") + raise ValueError( + f"No private keys for {self.email}. Encryption is on for this " + "client, but it holds no key pair. Keys are created at login, " + "so log in again to create them." + ) return self._private_keys def _ensure_peer(self, email: str) -> Peer: peer = self.get_cached_peer(email) if peer is None: - raise ValueError(f"No cached peer for {email}") + raise SyftPeerNotReadyError( + email, + cause=f"No cached peer for {email}.", + remedy=( + "Run client.sync() to refresh your peers, or " + f"client.add_peer('{email}') if you have not added them." + ), + ) return peer def _ensure_peer_bundle(self, email: str) -> dict: peer = self._ensure_peer(email) if peer.public_encryption_bundle is None: - raise ValueError(f"No public encryption bundle for {email}") + cause, remedy = self._missing_bundle_reason(peer) + raise SyftPeerNotReadyError(email, cause=cause, remedy=remedy) return peer.public_encryption_bundle + @staticmethod + def _missing_bundle_reason(peer: Peer) -> tuple[str, str]: + """The cause and the remedy for a peer that has no encryption bundle. + + A bundle is only stored for a peer who is accepted (or requested by + you) and who has already published one, so the four states below are + the four reasons the bundle can be missing. + """ + email = peer.email + if peer.state == PeerState.REQUESTED_BY_PEER: + return ( + f"{email} asked to peer with you, but you have not accepted yet.", + f"Run client.approve_peer_request('{email}').", + ) + if peer.state == PeerState.REJECTED: + return ( + f"You rejected {email}, so no encryption bundle was kept.", + f"Run client.add_peer('{email}') to peer with them again.", + ) + if peer.state == PeerState.REQUESTED_BY_ME: + return ( + f"You asked to peer with {email}, but they have not accepted yet.", + "Wait for them to accept, then run client.sync().", + ) + return ( + f"{email} is an accepted peer, but you do not have their public " + "encryption bundle. Either they have not published one yet, or you " + "have not synced since they did.", + "Run client.sync(). If it continues, the peer may not have " + "encryption enabled.", + ) + # ========== Crypto methods ========== def generate_keys(self) -> None: diff --git a/tests/unit/test_encryption.py b/tests/unit/test_encryption.py index cfc839d871f..9d0fb18200f 100644 --- a/tests/unit/test_encryption.py +++ b/tests/unit/test_encryption.py @@ -3,7 +3,8 @@ import pytest -from syft.sync.peers.peer import Peer +from syft.sync.peers.exceptions import SyftPeerNotReadyError +from syft.sync.peers.peer import Peer, PeerState from syft.sync.peers.peer_store import PeerStore from syft.sync.syftbox_manager import SyftboxManager from tests.unit.test_sync_manager import path_for_job @@ -49,22 +50,62 @@ def test_encrypt_decrypt_roundtrip(): def test_encrypt_without_keys_raises(): ps = PeerStore(email="alice@example.com", use_encryption=True) ps.add_peer(Peer(email="bob@example.com")) - with pytest.raises(ValueError, match="No private keys"): + with pytest.raises(ValueError, match="Keys are created at login"): ps.encrypt("bob@example.com", b"data") -def test_encrypt_without_peer_bundle_raises(): +@pytest.mark.parametrize( + "state, cause, remedy", + [ + ( + PeerState.ACCEPTED, + "accepted peer, but you do not have their public", + "Run client.sync()", + ), + ( + PeerState.REQUESTED_BY_ME, + "they have not accepted yet", + "Wait for them to accept", + ), + ( + PeerState.REQUESTED_BY_PEER, + "you have not accepted yet", + "client.approve_peer_request('bob@example.com')", + ), + ( + PeerState.REJECTED, + "You rejected bob@example.com", + "client.add_peer('bob@example.com')", + ), + ], +) +def test_encrypt_without_peer_bundle_names_the_cause(state, cause, remedy): + """A missing bundle means one of four things; the message must say which. + + Naming neither the cause nor the remedy is what left data owners stuck on + this error with nothing to act on. + """ ps = PeerStore(email="alice@example.com", use_encryption=True) ps.generate_keys() - ps.add_peer(Peer(email="bob@example.com")) - with pytest.raises(ValueError, match="No public encryption bundle"): + ps.add_peer(Peer(email="bob@example.com", state=state)) + + with pytest.raises(SyftPeerNotReadyError) as exc: ps.encrypt("bob@example.com", b"data") + assert cause in exc.value.cause + assert remedy in exc.value.remedy + assert str(exc.value) == f"{exc.value.cause} {exc.value.remedy}" + + +def test_peer_not_ready_error_is_a_value_error(): + """Callers that catch ValueError keep working.""" + assert issubclass(SyftPeerNotReadyError, ValueError) + def test_try_decrypt_no_keys(): ps = PeerStore(email="alice@example.com", use_encryption=True) data = b"some unencrypted data" - with pytest.raises(ValueError, match="No private keys"): + with pytest.raises(ValueError, match="Keys are created at login"): ps.decrypt("bob@example.com", data) == data @@ -72,8 +113,10 @@ def test_try_decrypt_no_peer_bundle(): ps = PeerStore(email="alice@example.com", use_encryption=True) ps.generate_keys() data = b"some unencrypted data" - with pytest.raises(ValueError, match="No cached peer for"): - ps.decrypt("bob@example.com", data) == data + with pytest.raises(SyftPeerNotReadyError) as exc: + ps.decrypt("bob@example.com", data) + assert "No cached peer for bob@example.com" in exc.value.cause + assert "client.add_peer('bob@example.com')" in exc.value.remedy def test_try_decrypt_invalid_envelope(): diff --git a/tests/unit/test_package_logging.py b/tests/unit/test_package_logging.py new file mode 100644 index 00000000000..cabef30476d --- /dev/null +++ b/tests/unit/test_package_logging.py @@ -0,0 +1,37 @@ +"""Each package must configure its own logger when imported on its own. + +`syft/__init__.py` configures only the `syft` logger. The other packages are +sibling top-level namespaces, so without this they inherit the root logger — +no handler, level WARNING — and every logger.info/warning/error call in them +is dropped, swallowed tracebacks included. + +Each case runs in its own interpreter: the point is that importing the package +alone is enough, with no dependence on `syft` being imported first. +""" + +import subprocess +import sys + +import pytest + +PACKAGES = ["syft", "syft_job", "syft_rds", "syft_enclaves", "syft_bg"] + +PROBE = """ +import importlib, logging, sys +importlib.import_module({name!r}) +logger = logging.getLogger({name!r}) +print(logger.getEffectiveLevel(), bool(logger.handlers)) +""" + + +@pytest.mark.parametrize("package", PACKAGES) +def test_package_logger_is_audible(package): + result = subprocess.run( + [sys.executable, "-c", PROBE.format(name=package)], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + level, has_handler = result.stdout.strip().splitlines()[-1].split() + assert int(level) == 20, f"{package} effective level is {level}, want INFO (20)" + assert has_handler == "True", f"{package} logger has no handler" diff --git a/uv.lock b/uv.lock index 375dd1ae181..dcecba1608a 100644 --- a/uv.lock +++ b/uv.lock @@ -1535,10 +1535,10 @@ resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.12.*' and sys_platform == 'win32'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform == 'win32'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.11.*' and sys_platform == 'emscripten'", + "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ @@ -2977,10 +2977,10 @@ resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.12.*' and sys_platform == 'win32'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform == 'win32'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.11.*' and sys_platform == 'emscripten'", + "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ @@ -4265,10 +4265,10 @@ resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.12.*' and sys_platform == 'win32'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform == 'win32'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.11.*' and sys_platform == 'emscripten'", + "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] sdist = { url = "https://files.pythonhosted.org/packages/aa/2a/9618a122aeb2a169a28b03889a2995fe297588964333d4a7d67bdf46e147/rpds_py-2026.6.3.tar.gz", hash = "sha256:1cebd1337c242e4ec2293e541f712b2da849b29f48f0c293684b71c0632625d4", size = 64051, upload-time = "2026-06-30T07:17:53.009Z" } @@ -4469,7 +4469,6 @@ dependencies = [ { name = "rich" }, { name = "syft-crypto-python" }, { name = "syft-dataset" }, - { name = "syft-migration" }, { name = "syft-permissions" }, { name = "syft-perms" }, ] @@ -4513,7 +4512,6 @@ requires-dist = [ { name = "rich", specifier = ">=13.0.0" }, { name = "syft-crypto-python", specifier = ">=0.1.2b2" }, { name = "syft-dataset", editable = "packages/syft-datasets" }, - { name = "syft-migration", editable = "packages/syft-migration" }, { name = "syft-permissions", editable = "packages/syft-permissions" }, { name = "syft-perms", editable = "packages/syft-perms" }, ] @@ -4548,7 +4546,7 @@ test = [ [[package]] name = "syft-bg" -version = "0.3.13" +version = "0.3.12" source = { editable = "packages/syft-bg" } dependencies = [ { name = "click" }, @@ -4594,7 +4592,7 @@ wheels = [ [[package]] name = "syft-dataset" -version = "0.1.22" +version = "0.1.21" source = { editable = "packages/syft-datasets" } dependencies = [ { name = "pyyaml" },