diff --git a/MIGRATING.md b/MIGRATING.md index 6c93ad3..b673bf5 100644 --- a/MIGRATING.md +++ b/MIGRATING.md @@ -17,7 +17,7 @@ line a day to stderr (the web server's error log): | `nea-errors` | errors are VOTable documents; a table not in TAP_SCHEMA returns 400 | plain text; 403 | | `nea-tables` | `/tables` uses VODataService `` elements | per-column ``, ``, ``, ``, `` | | `nea-votable` | VOTable results are `application/xml`, non-char FIELDs carry no DESCRIPTION (units are still written) | `text/xml`, with descriptions on every FIELD | -| `nea-uws` | job documents use `executionDuration`, `application/xml`, no `Z` on `destruction`; the 303 that starts a job (`PHASE=RUN`) carries no `Content-Type`, `Content-Length` or `Connection` lines, because the response ends when the CGI exits; `/async//executionDuration` (camelCase) is accepted as well as `/executionduration` | `executionduration`, `text/xml`, `Z`; the 303 carries `Content-Type: text/plain`, `Content-Length` and `Connection: close`; only `/executionduration` is a valid key | +| `nea-uws` | job documents are `application/xml`, no `Z` on `destruction`; the 303 that starts a job (`PHASE=RUN`) carries no `Content-Type`, `Content-Length` or `Connection` lines, because the response ends when the CGI exits; `/async//executionDuration` (camelCase) is accepted as well as `/executionduration`; job URLs join `HTTP_URL` and `CGI_PGM` with `/` as written, so `CGI_PGM=/TAP` gives `//TAP` | `text/xml`, `Z`; the 303 carries `Content-Type: text/plain`, `Content-Length` and `Connection: close`; only `/executionduration` is a valid key; job URLs have one `/` between `HTTP_URL` and `CGI_PGM` | ## Moving to the 3.x layout @@ -48,22 +48,27 @@ Compatibility mode reproduces 1.2's responses, with these exceptions: - Errors raised before `TAP.conf` is read (a missing or broken `TAP.conf`, a malformed request path) use the 3.x error format, since the mode isn't known yet. -- `/availability` and `/capabilities` create a workspace directory per - request, so workspace cleanup must cover them. - The async job document escapes quotes in the query text as `'`; it parses to the same XML. -- A job's `status.xml` written under one mode is read correctly under the - other (either spelling of the duration element), so switching `nea-uws` off - doesn't break jobs in flight. +- Both modes write the job document's duration as `executionDuration` (the + UWS name) and read either spelling, so jobs written by an earlier release + or under the other mode read correctly. +- Both modes follow UWS where 1.2 did not: a job can be deleted + (`POST ACTION=DELETE` or HTTP `DELETE`, answered with 303 to the job list), + an unknown job id is 404, `GET` on the job list returns an empty + `` instead of creating a job, and an unset job value (such as + `/destruction` before the job runs) is empty text instead of `None`. `COMPAT` is read only from `[WEB]`; placing it in any other section, including a 1.x `[webserver]`, is a configuration error. ## Release plan -- **3.1.0**: compatibility mode, the 1.x reader, this guide. -- **3.x**: where the shared behavior is to change (for example - `executionDuration` casing and VOTable errors, which IVOA standards - require), the release notes say so before any default changes. +- **3.1.0**: compatibility mode, the 1.x reader, this guide. The default + job document now spells the duration `executionDuration`, as UWS + requires, which retires that part of `nea-uws`. +- **3.x**: where the shared behavior is to change (for example VOTable + errors, which IVOA standards require), the release notes say so before + any default changes. - **4.0.0**: compatibility mode, the `COMPAT` key and the 1.x reader are removed; a 1.x `TAP.conf` then fails at startup with a pointer here. diff --git a/TAP/compat.py b/TAP/compat.py index 965f3f6..96e83ea 100644 --- a/TAP/compat.py +++ b/TAP/compat.py @@ -169,9 +169,15 @@ def field_description(active, desc, coltype): # --- nea-uws ---------------------------------------------------------------- -def duration_element(active): - """UWS names it executionDuration; the shared line writes it lowercase.""" - return 'executionDuration' if 'nea-uws' in active else 'executionduration' +def service_url(active, httpurl, cgipgm): + """The service URL that job URLs extend. + + NEA's carry a double slash (HTTP_URL + '/' + CGI_PGM=/TAP); the shared + line joins the parts with one. + """ + if 'nea-uws' in active: + return httpurl + '/' + cgipgm + return httpurl.rstrip('/') + '/' + cgipgm.strip('/') def uws_key(job, key): @@ -199,7 +205,8 @@ def duration_value(job): """The duration in a parsed status.xml, whichever spelling wrote it. A job written under one mode may be read under the other (jobs live up - to four days), so reading tolerates both; writing uses duration_element. + to four days, and 3.1.0 and earlier wrote it lowercase), so reading + tolerates both; writing uses the UWS spelling. """ return job.get('uws:executionDuration', job.get('uws:executionduration')) diff --git a/TAP/tap.py b/TAP/tap.py index c77b74a..b262862 100755 --- a/TAP/tap.py +++ b/TAP/tap.py @@ -9,6 +9,8 @@ import logging import math import os +import re +import shutil import signal import sys import tempfile @@ -170,7 +172,13 @@ def __run__(self, **kwargs): # { tap.init() # - if('debug' in self.form): + # A request with no form fields (an HTTP DELETE) has no list, and + # FieldStorage raises 'not indexable' on any lookup into it. + self.fields = self.form.keys() if self.form.list is not None else [] + self.request_method = os.environ.get('REQUEST_METHOD', '').upper() + self.action = '' + + if('debug' in self.fields): self.debug = 1 if(self.debug): @@ -227,7 +235,7 @@ def __run__(self, **kwargs): self.uwsheader = '' - for key in self.form: + for key in self.fields: if self.debug: logging.debug(f' key: {key:<15} val: {self.form[key].value:s}') @@ -267,6 +275,9 @@ def __run__(self, **kwargs): if(key.lower() == 'token'): self.token = self.form[key].value + if(key.lower() == 'action'): + self.action = self.form[key].value.strip().upper() + self.nparam = len(self.param) if 'format' in self.param: @@ -526,27 +537,6 @@ def __run__(self, **kwargs): - # Special case: HTTP DELETE. We only need - # the REQUEST_METHOD and PATH_INFO environment - # variable. The first must be DELETE and the - # second is the directory of th TAP job we wish - # to delete. - # - - # if("REQUEST_METHOD" in os.environ): - # self.request_method = os.environ["REQUEST_METHOD"] - # else: - # self.request_method = '' - # - # if self.request_method == 'DELETE': - # - # delete_dir = self.workdir + self.pathinfo - # - # if self.debug: - # logging.debug('') - # logging.debug(f'DELETE: {delete_dir:s}') - - # # Initialize statdict dict # @@ -588,6 +578,15 @@ def __run__(self, **kwargs): #} end VOSI static endpoints # + # + # GET on the job list must not create a job: clients follow the + # redirect that answers a job deletion with a GET here. + # + if ((self.tapcontext == 'async') and (self.getstatus == 0) and + (self.querykey == 0) and (len(self.param['phase']) == 0) and + (self.request_method != 'POST')): + self.__printJobList__() + # # sync or async without input workspace id: make workspace, # otherwise retrieve workspace from getstatus id @@ -656,6 +655,15 @@ def __run__(self, **kwargs): # Retrieve workspace from id # + # + # A job id names a directory that tempfile.mkdtemp made under + # /TAP; anything else ('..', '.') would reach outside + # it, and a deletion removes what the id names. + # + if not re.fullmatch(r'tap_[A-Za-z0-9_]+', self.id): + self.__printError__('votable', 'Job not found: ' + self.id, + errcode='404') + self.workspace = self.id self.userWorkdir = self.workdir + '/TAP/' + self.workspace @@ -663,11 +671,18 @@ def __run__(self, **kwargs): # # check if workspace exists # - isExist = os.path.exists(self.userWorkdir) + isExist = os.path.isdir(self.userWorkdir) if (isExist == 0): - self.msg = 'work directory based on input jobid does not exist.' - self.__printError__('votable', self.msg, errcode='500') + self.__printError__('votable', 'Job not found: ' + self.id, + errcode='404') + + if ((self.tapcontext == 'async') and + (self.statuskey.strip('/') == '') and + ((self.request_method == 'DELETE') or + ((self.request_method == 'POST') and + (self.action == 'DELETE')))): + self.__deleteJob__() # # } end of retrieve workspace @@ -712,7 +727,8 @@ def __run__(self, **kwargs): self.statustbl = 'status.xml' self.statuspath = self.userWorkdir + '/' + self.statustbl - self.statusurl = self.httpurl + '/' + self.cgipgm + \ + self.statusurl = \ + compat.service_url(self.compat, self.httpurl, self.cgipgm) + \ '/' + self.tapcontext + '/' + self.workspace if self.debug: @@ -2284,7 +2300,11 @@ def __getStatus__(self, workdir, workspace, key, param, **kwargs): or (key == 'jobId') or (key == 'runId')): - retval = job[keystr] + # xmltodict reads an empty element as None and a nil one + # as a dict; both are an unset value, which is empty text. + retval = job.get(keystr) + if (retval is None) or isinstance(retval, dict): + retval = '' outstr = retval elif((key == 'ownerId') or (key == 'quote')): @@ -2756,6 +2776,69 @@ def __printSyncResponse__(self, status, msg, resulturl, format, **kwargs): # + def __printJobList__(self, **kwargs): + + # + # UWS job list (GET /async). Jobs carry no owner, so the list + # shows none of them rather than everyone's. + # + + body = ('\n' + '\n') + + sys.stdout.write('HTTP/1.1 200 OK\r\n') + sys.stdout.write(f'Content-Type: {compat.uws_content_type(self.compat)}\r\n') + sys.stdout.write(f'Content-Length: {len(body.encode("utf-8"))}\r\n') + sys.stdout.write('Connection: close\r\n\r\n') + sys.stdout.write(body) + sys.stdout.flush() + sys.exit() + + + def __deleteJob__(self, **kwargs): + + # + # UWS job deletion (POST ACTION=DELETE, or HTTP DELETE): stop the + # job if it is running, remove its workspace, and send the client + # to the job list. The job id was checked against the workspace + # name pattern before userWorkdir was built from it. + # + + try: + with open(self.userWorkdir + '/status.xml', 'r') as fp: + job = xmltodict.parse(fp.read())['uws:job'] + + if (job.get('uws:phase') == 'EXECUTING'): + pid = int(job.get('uws:runId') or 0) + if (pid > 0) and (pid != os.getpid()): + os.kill(pid, signal.SIGKILL) + + except Exception as e: + + # A job whose status can't be read or whose process is gone + # is still deleted. + if self.debug: + logging.debug(f'deleteJob: {str(e):s}') + + try: + shutil.rmtree(self.userWorkdir) + + except OSError as e: + self.__printError__('votable', 'Failed to delete job ' + + self.workspace + ': ' + str(e), errcode='500') + + joblist = compat.service_url(self.compat, self.httpurl, self.cgipgm) + \ + '/async' + + sys.stdout.write('HTTP/1.1 303 See Other\r\n') + sys.stdout.write(f'Location: {joblist}\r\n') + sys.stdout.write('Content-Length: 0\r\n') + sys.stdout.write('Connection: close\r\n\r\n') + sys.stdout.flush() + sys.exit() + + def __printAsyncResponse__(self, statusurl, **kwargs): # @@ -3051,8 +3134,8 @@ def __writeStatusMsg__(self, statuspath, statdict, param, **kwargs): fp.write(f" {statdict['endtime']:s}\n") - el = compat.duration_element(self.compat) - fp.write(f" {statdict['duration']:d}\n") + fp.write(f" {statdict['duration']:d}" + "\n") if (statdict['destruction'] is None): fp.write(' \n') diff --git a/tests/conftest.py b/tests/conftest.py index e2e9b20..10f57cf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -101,6 +101,9 @@ def do_GET(self): def do_POST(self): self._run_cgi() + def do_DELETE(self): + self._run_cgi() + def _run_cgi(self): # Parse PATH_INFO and QUERY_STRING from self.path. from urllib.parse import urlsplit diff --git a/tests/test_compat_http.py b/tests/test_compat_http.py index 970c29a..29672f5 100644 --- a/tests/test_compat_http.py +++ b/tests/test_compat_http.py @@ -119,7 +119,7 @@ def test_uws_is_neas_form(legacy_tap_server): assert camel.text == dur.text -def test_uws_unchanged_in_default_mode(tap_server): +def test_uws_in_default_mode(tap_server): run, url = _completed_job(tap_server) assert run.headers["Content-Type"] == "text/plain" assert int(run.headers["Content-Length"]) == len(run.content) @@ -127,7 +127,7 @@ def test_uws_unchanged_in_default_mode(tap_server): r = requests.get(url, timeout=30) assert r.headers["Content-Type"] == "text/xml" job = ET.fromstring(r.text) - assert job.find(UWS + "executionduration") is not None + assert job.find(UWS + "executionDuration") is not None # UWS spelling, as of 3.1.0 assert job.find(UWS + "destruction").text.endswith("Z") @@ -140,7 +140,7 @@ def test_camelcase_duration_key_unchanged_in_default_mode(tap_server): def _respell_duration(fixture_root, url, old, new): - """Rewrite one job's status.xml as if the other mode had written it.""" + """Rewrite one job's status.xml as an earlier release would have written it.""" jobid = url.rstrip("/").rsplit("/", 1)[1] paths = list(fixture_root.glob("workdir*/**/%s/status.xml" % jobid)) assert len(paths) == 1, paths @@ -158,15 +158,18 @@ def _read_back(url): assert phase.text.strip() == "COMPLETED", phase.text[:300] -def test_status_written_by_default_mode_reads_in_compat(legacy_tap_server, fixture_root): +# Jobs live four days, so a job written lowercase by an earlier release can +# be read after an upgrade, in either mode. + +def test_lowercase_status_reads_in_compat(legacy_tap_server, fixture_root): _, url = _completed_job(legacy_tap_server) _respell_duration(fixture_root, url, "executionDuration", "executionduration") _read_back(url) -def test_status_written_by_compat_reads_in_default_mode(tap_server, fixture_root): +def test_lowercase_status_reads_in_default_mode(tap_server, fixture_root): _, url = _completed_job(tap_server) - _respell_duration(fixture_root, url, "executionduration", "executionDuration") + _respell_duration(fixture_root, url, "executionDuration", "executionduration") _read_back(url) diff --git a/tests/test_compat_units.py b/tests/test_compat_units.py index 7059818..a7a94c8 100644 --- a/tests/test_compat_units.py +++ b/tests/test_compat_units.py @@ -34,8 +34,6 @@ def test_field_description_blanks_only_non_char_columns(): def test_uws_functions(): - assert compat.duration_element(compat.ALL) == 'executionDuration' - assert compat.duration_element(compat.NONE) == 'executionduration' assert compat.uws_key({'uws:executionDuration': '1'}, 'executionduration') == 'executionDuration' assert compat.uws_key({'uws:executionduration': '1'}, 'executionduration') == 'executionduration' assert compat.uws_key({'uws:executionDuration': '1'}, 'phase') == 'phase' diff --git a/tests/test_uws_conformance.py b/tests/test_uws_conformance.py new file mode 100644 index 0000000..e8b3fda --- /dev/null +++ b/tests/test_uws_conformance.py @@ -0,0 +1,142 @@ +"""UWS behaviors taplint checks (E-UWS-DENO, -HTDE, -JDED, -JDDE) and what +deleting a job needs to be safe.""" +from __future__ import annotations + +import http.client +import os +import xml.etree.ElementTree as ET +from pathlib import Path +from urllib.parse import urlsplit + +import pytest +import requests +from _helpers import tap_async_url + +from TAP import compat + +UWS = "{http://www.ivoa.net/xml/UWS/v1.0}" +QUERY = "SELECT TOP 2 table_name FROM TAP_SCHEMA.tables" + + +def _submit(server): + r = requests.post(tap_async_url(server), data={"query": QUERY, "lang": "ADQL", "format": "csv"}, + allow_redirects=False, timeout=60) + assert r.status_code == 303, r.text + return r.headers["Location"] + + +def _job(url): + r = requests.get(url, timeout=60) + assert r.status_code == 200, r.text + return ET.fromstring(r.content) + + +def _text(url, key): + r = requests.get(f"{url}/{key}", timeout=60) + assert r.status_code == 200, r.text + return r.text.strip() + + +def _jobs_dir(fixture_root: Path) -> Path: + return fixture_root / "workdir" / "TAP" + + +@pytest.mark.parametrize("phase", ["PENDING", "COMPLETED"]) +def test_job_document_agrees_with_its_resources(tap_server, phase): + """JDED, JDDE: the job document and /executionduration, /destruction match.""" + url = _submit(tap_server) + if phase == "COMPLETED": + requests.post(url + "/phase", data={"PHASE": "RUN"}, allow_redirects=False, timeout=60) + for _ in range(60): + if _text(url, "phase") == "COMPLETED": + break + job = _job(url) + duration = job.find(UWS + "executionDuration") + assert duration is not None, "UWS names the element executionDuration" + assert (duration.text or "") == _text(url, "executionduration") + assert (job.find(UWS + "destruction").text or "") == _text(url, "destruction") + + +@pytest.mark.parametrize("how", ["POST ACTION=DELETE", "HTTP DELETE"]) +def test_deleted_job_is_gone(tap_server, fixture_root, how): + """DENO, HTDE: deleting answers 303 to the job list; the job is then 404.""" + url = _submit(tap_server) + jobid = url.rstrip("/").rsplit("/", 1)[1] + assert (_jobs_dir(fixture_root) / jobid).is_dir() + if how == "HTTP DELETE": + r = requests.delete(url, allow_redirects=False, timeout=60) + else: + r = requests.post(url, data={"ACTION": "DELETE"}, allow_redirects=False, timeout=60) + assert r.status_code == 303, r.text + assert r.headers["Location"].rstrip("/").endswith("/async") + assert requests.get(url, timeout=60).status_code == 404 + assert not (_jobs_dir(fixture_root) / jobid).exists() + + +def test_completed_job_can_be_deleted(tap_server): + url = _submit(tap_server) + requests.post(url + "/phase", data={"PHASE": "RUN"}, allow_redirects=False, timeout=60) + for _ in range(60): + if _text(url, "phase") == "COMPLETED": + break + r = requests.delete(url, allow_redirects=False, timeout=60) + assert r.status_code == 303, r.text + assert requests.get(url, timeout=60).status_code == 404 + + +def test_unknown_job_is_404(tap_server): + url = tap_async_url(tap_server) + "/tap_doesnotexist" + assert requests.get(url, timeout=60).status_code == 404 + assert requests.delete(url, allow_redirects=False, timeout=60).status_code == 404 + + +def _raw_status(server, method, path, body=b""): + """Send path as is: requests would drop '.' and '..' segments first.""" + parts = urlsplit(server) + conn = http.client.HTTPConnection(parts.hostname, parts.port, timeout=60) + headers = {"Content-Type": "application/x-www-form-urlencoded"} if body else {} + conn.request(method, path, body=body, headers=headers) + status = conn.getresponse().status + conn.close() + return status + + +@pytest.mark.parametrize("jobid", ["..", ".", "TAP", "tap_x.y", "notajob"]) +def test_job_ids_outside_the_job_namespace_are_404(tap_server, fixture_root, jobid): + """A job id names a directory under /TAP; nothing else may be reached.""" + before = sorted(os.listdir(fixture_root / "workdir")) + path = urlsplit(tap_async_url(tap_server)).path + "/" + jobid + assert _raw_status(tap_server, "DELETE", path) == 404 + assert _raw_status(tap_server, "POST", path, b"ACTION=DELETE") == 404 + assert _raw_status(tap_server, "GET", path) == 404 + assert sorted(os.listdir(fixture_root / "workdir")) == before + assert _jobs_dir(fixture_root).is_dir() + + +def test_job_list_does_not_create_a_job(tap_server, fixture_root): + """Clients follow the delete redirect with a GET on the job list.""" + before = set(os.listdir(_jobs_dir(fixture_root))) + r = requests.get(tap_async_url(tap_server), allow_redirects=False, timeout=60) + assert r.status_code == 200, r.text + assert ET.fromstring(r.content).tag == UWS + "jobs" + assert set(os.listdir(_jobs_dir(fixture_root))) == before + + +@pytest.mark.parametrize("httpurl, cgipgm", [ + ("https://host.org", "/TAP"), ("https://host.org/", "TAP"), ("https://host.org", "TAP"), + ("https://host.org/", "/TAP/"), +]) +def test_job_urls_have_one_slash_between_parts(httpurl, cgipgm): + assert compat.service_url(compat.NONE, httpurl, cgipgm) == "https://host.org/TAP" + + +def test_nea_keeps_its_job_url_form(): + # NEA's job URLs carry the double slash today (HTTP_URL + '/' + CGI_PGM=/TAP). + assert compat.service_url(compat.ALL, "https://host.org", "/TAP") == "https://host.org//TAP" + + +def test_delete_works_in_compat_mode(legacy_tap_server, fixture_root): + url = _submit(legacy_tap_server) + r = requests.delete(url, allow_redirects=False, timeout=60) + assert r.status_code == 303, r.text + assert requests.get(url, timeout=60).status_code == 404