diff --git a/MIGRATING.md b/MIGRATING.md new file mode 100644 index 0000000..417b01f --- /dev/null +++ b/MIGRATING.md @@ -0,0 +1,69 @@ +# Migrating from the 1.x TAP.conf layout and NEA compatibility mode + +nexsciTAP 3.1 runs deployments that still use the 1.x `TAP.conf` layout +(`[webserver]`, a section named by `DBMS=`, top-level `ADQL_*` keys). A 1.x +config is translated to the 3.x layout when it's read, and it turns on +**compatibility mode**: the responses that deployment sent under nexsciTAP +1.2. Compatibility mode is temporary. While it's on, the service writes one +line a day to stderr (the web server's error log): + + nexsciTAP: compatibility mode active (...); see MIGRATING.md + +## Compatibility behaviors + +| Name | With it (1.2 behavior) | Without it (3.x behavior) | +|---|---|---| +| `nea-vosi-headers` | `/availability` and `/capabilities` send a status line and headers | the document only | +| `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 | + +## Moving to the 3.x layout + +1. Translate the config, keeping every behavior: in `[WEB]`, list them all. + + | 1.x | 3.x | + |---|---| + | `[webserver]` `TAP_WORKDIR`, `TAP_WORKURL`, `HTTP_URL`, `HTTP_PORT`, `CGI_PGM`, `ArraySize`, `INFOMSG` | `[WEB]`, same keys | + | `[webserver]` `DBMS` | `[DBMS]` `DBMS` | + | the `[oracle]` / `[sqlite3]` section's keys | `[DBMS]`, same keys | + | `[webserver]` `COOKIENAME`, `RACOL`, `DECCOL` | `[DBMS]`, same keys | + | `ADQL_MODE`, `ADQL_LEVEL`, `ADQL_XCOL`, `ADQL_YCOL`, `ADQL_ZCOL`, `ADQL_COLNAME`, `ADQL_ENCODING` | `[SPTIND]` `MODE`, `LEVEL`, `XCOL`, `YCOL`, `ZCOL`, `COLNAME`, `ENCODING` | + + ```ini + [WEB] + COMPAT = nea-vosi-headers, nea-errors, nea-tables, nea-votable, nea-uws + ``` + Responses don't change. + +2. Remove names from `COMPAT` one at a time, each as an announced change for + that service's users. +3. When `COMPAT` is empty, remove the key. + +## Known differences from 1.2 + +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. + +`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. +- **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 new file mode 100644 index 0000000..c3c87d6 --- /dev/null +++ b/TAP/compat.py @@ -0,0 +1,255 @@ +"""NEA compatibility mode: NEA's pre-3.x responses on the shared code line. + +Compat mode is on when TAP.conf uses the 1.x layout ([webserver], no [WEB]), +which implies every behavior below, or when [WEB] lists COMPAT names. Each +function returns what the default code does unless its behavior is active, +so call sites hold no conditionals, and retiring compat means deleting this +module and compat_vositables.py. See MIGRATING.md for the plan. +""" +import datetime +import html +import os +import sys + +import configobj + +NAMES = ('nea-vosi-headers', 'nea-errors', 'nea-tables', 'nea-votable', 'nea-uws') +ALL = frozenset(NAMES) +NONE = frozenset() + +# 1.x [webserver] keys that 3.x reads from [WEB]. +_WEB_KEYS = ('TAP_WORKDIR', 'TAP_WORKURL', 'HTTP_URL', 'HTTP_PORT', 'CGI_PGM', + 'ArraySize', 'INFOMSG') +# 1.x [webserver] keys that 3.x reads from the connection section. +_DB_KEYS_FROM_WEB = ('COOKIENAME', 'RACOL', 'DECCOL') +# 1.x ADQL_* keys and their 3.x [SPTIND] names. +_SPTIND_KEYS = {'ADQL_MODE': 'MODE', 'ADQL_LEVEL': 'LEVEL', 'ADQL_XCOL': 'XCOL', + 'ADQL_YCOL': 'YCOL', 'ADQL_ZCOL': 'ZCOL', + 'ADQL_COLNAME': 'COLNAME', 'ADQL_ENCODING': 'ENCODING'} + + +class CompatConfigError(Exception): + pass + + +def parse_names(value): + """A COMPAT value (configobj gives str or list) -> frozenset of names.""" + if value is None: + return NONE + items = [value] if isinstance(value, str) else list(value) + names = frozenset(v.strip() for v in items if v.strip()) + unknown = sorted(names - ALL) + if unknown: + raise CompatConfigError('unknown COMPAT name(s) %s; valid names: %s' + % (', '.join(unknown), ', '.join(NAMES))) + return names + + +def _find(conf, key): + """A 1.x key wherever configobj filed it: [webserver] first, then any section. + + Keys written after a section header belong to that section, so ADQL_* + lines placed after [oracle] land in [oracle]. + """ + if key in conf['webserver']: + return conf['webserver'][key] + for name in conf.sections: + if key in conf[name]: + return conf[name][key] + return None + + +def translate_legacy(conf): + """A 1.x TAP.conf ([webserver], [], ADQL_*) as an equivalent 3.x one.""" + web = conf['webserver'] + dbms = web.get('DBMS', '') + if dbms not in conf: + raise CompatConfigError('1.x TAP.conf names DBMS=%s but has no [%s] section' + % (dbms, dbms)) + new = configobj.ConfigObj() + new['WEB'] = {k: web[k] for k in _WEB_KEYS if k in web} + db = {'DBMS': dbms} + db.update({k: v for k, v in conf[dbms].items() if not k.startswith('ADQL_')}) + for k in _DB_KEYS_FROM_WEB: + if k in web: + db[k] = web[k] + if dbms == 'sqlite3': + db.setdefault('TAP_SCHEMA_FILE', 'TAP_SCHEMA') # 3.x's SQLite ATTACH name + new['DBMS'] = db + sptind = {} + for old, current in _SPTIND_KEYS.items(): + value = _find(conf, old) + if value is not None: + sptind[current] = value + new['SPTIND'] = sptind + return new + + +def load_config(conf): + """(ConfigObj as read) -> (ConfigObj in 3.x layout, active compat names).""" + for name in conf.sections: + if name != 'WEB' and 'COMPAT' in conf[name]: + raise CompatConfigError('COMPAT belongs in [WEB], found in [%s]' % name) + if 'WEB' not in conf and 'webserver' in conf: + return translate_legacy(conf), ALL + web = conf['WEB'] if 'WEB' in conf else {} + return conf, parse_names(web.get('COMPAT')) + + +# --- nea-vosi-headers ------------------------------------------------------- + +def vosi_head(active): + """Status line and headers NEA's nph- CGI writes before a VOSI document.""" + if 'nea-vosi-headers' not in active: + return '' + return ('HTTP/1.1 200 OK\r\nContent-type: application/xml\r\n' + 'Connection: close\r\n\r\n') + + +def vosi_tail(active): + """NEA ends its VOSI documents with one more CRLF.""" + return '\r\n' if 'nea-vosi-headers' in active else '' + + +# --- nea-errors ------------------------------------------------------------- + +def table_denied_status(active): + """Status for a table outside TAP_SCHEMA: NEA answers 400, the shared line 403.""" + return '400' if 'nea-errors' in active else '403' + + +def error_document(active, errmsg, errcode): + """NEA's error response: a VOTable error document whatever the format. + + Returns the whole response, or None when nea-errors is inactive. Markup + is escaped so the document stays well-formed; quotes are not, matching + NEA's bytes for ordinary messages. + """ + if 'nea-errors' not in active: + return None + return ('HTTP/1.1 %s ERROR\r\n' % errcode + + 'Content-type: application/xml\r\n\r\n' + + '\n' + + '\n' + + '\n' + + '\n' + + html.escape(str(errmsg), quote=False) + '\n' + + '\n' + + '\n' + + '\n') + + +# --- nea-votable ------------------------------------------------------------ + +def votable_content_type(active): + """Content type for VOTable results: NEA sends application/xml.""" + return 'application/xml' if 'nea-votable' in active else 'text/xml' + + +def field_description(active, desc, coltype): + """NEA's VOTables carry no DESCRIPTION for non-char columns. + + The C writer prints the plain form when a non-char column's + description is empty (writerecsmodule.c, non-char branch), so blanking it + here reproduces NEA's FIELDs without touching the C API. Char columns + already match. + """ + if 'nea-votable' in active and coltype != 'char': + return '' + return desc + + +# --- 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 uws_key(job, key): + """Element of a parsed status.xml that answers a job sub-resource key. + + The duration is spelled as the document spells it (see duration_value). + """ + if key == 'executionduration' and 'uws:executionDuration' in job: + return 'executionDuration' + return key + + +def uws_url_key(active, key): + """The job sub-resource key from the URL, as the code compares it. + + NEA's /async//executionDuration (UWS spelling) is accepted next to the + lowercase form pyvo uses; the shared line knows only the lowercase one. + """ + if 'nea-uws' in active and key == 'executionDuration': + return 'executionduration' + return key + + +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. + """ + return job.get('uws:executionDuration', job.get('uws:executionduration')) + + +def destruction_suffix(active): + """NEA writes the destruction time without a trailing 'Z'.""" + return '' if 'nea-uws' in active else 'Z' + + +def uws_content_type(active): + """Content type of the job document: NEA sends application/xml.""" + return 'application/xml' if 'nea-uws' in active else 'text/xml' + + +def async_submit_headers(active, body): + """Framing headers of the 303 that starts a job (PHASE=RUN). + + NEA sends none: the response ends when the CGI exits, because the + forked worker holds no stdout. Otherwise: Content-Type, the length of + `body` in bytes, and Connection: close. + """ + if 'nea-uws' in active: + return '' + return ('Content-Type: text/plain\r\nContent-Length: %d\r\n' + 'Connection: close\r\n' % len(body.encode('utf-8'))) + + +# --- nea-tables ------------------------------------------------------------- + +def vosi_tables_class(active): + """The /tables writer: NEA's (prod's vositables.py, verbatim) or the shared one.""" + if 'nea-tables' in active: + from TAP.compat_vositables import vosiTables + else: + from TAP.vositables import vosiTables + return vosiTables + + +# --- deprecation ------------------------------------------------------------ + +def warn_once_per_day(active, workdir, today=None, stream=None): + """While compat is active, one stderr line (Apache's error_log) per day. + + A stamp file /TAP/.compat-warned-, created exclusively, + marks the day, so concurrent requests warn once. Any failure is + swallowed: the warning must never fail a request. + """ + if not active: + return False + try: + day = (today or datetime.date.today()).isoformat() + stamp = os.path.join(workdir, 'TAP', '.compat-warned-' + day) + os.makedirs(os.path.dirname(stamp), exist_ok=True) + with open(stamp, 'x'): + pass + (stream or sys.stderr).write( + 'nexsciTAP: compatibility mode active (%s); see MIGRATING.md\n' + % ', '.join(sorted(active))) + return True + except Exception: + return False diff --git a/TAP/compat_vositables.py b/TAP/compat_vositables.py new file mode 100644 index 0000000..26f20b9 --- /dev/null +++ b/TAP/compat_vositables.py @@ -0,0 +1,1702 @@ +# ruff: noqa +# NEA compatibility (nea-tables): NEA prod's vositables.py, kept verbatim from +# commit bbc1108 so /tables matches NEA byte for byte. Selected by +# compat.vosi_tables_class(); delete with compat.py (see MIGRATING.md). + +# Copyright (c) 2020, Caltech IPAC. +# This code is released with a BSD 3-clause license. License information is at +# https://github.com/Caltech-IPAC/nexsciTAP/blob/master/LICENSE + + +import sys +import os +import logging + +import datetime + +import argparse + + +class vosiTables: +# +# { class vosiTables +# + + pid = os.getpid() + + debug = 0 + + debugfname = './vosi.debug' + status = '' + msg = '' + returnMsg = "" + + arraysize = 10000 + + # + # DD columns + # + + dd = None + + conn = None + dbtable = '' + + sql = '' + racol = '' + deccol = '' + + nfetch = 1000 + + outpath = '' + + #ntot = 0 + + format = 'votable' + maxrec = -1 + coldesc = 0 + + nschema = 0 + + schema_namearr = [] + schema_descarr = [] + + + def __init__(self, **kwargs): + + # + # { vosiTables.init method + # + + """ + vosiTables class generates the VOSI compatible XML outputs for TAP + query with endpoint TAP/tables + + Required keyword input parameters: + + connectInfo (python dictionary): + a python dictionary containing info needed to make + a "connection" to a DBMS. + + ourpath(char): output table file path + + Usage: + + vosi = vosiTables (\ + dbms=dbms, + dbserver=dbserver, \ + userid=userid, \ + password=password, \ + outpath=outpath, \ + debug=debug) + """ + + self.dbms = '' + self.dbserver = '' + self.userid = '' + self.password = '' + self.outpath = '' + self.debug = 0 + + self.db = '' + self.tap_schema = '' + + if('debug' in kwargs): + self.debug = kwargs['debug'] + + if self.debug: + logging.debug('') + logging.debug('Enter vosiTables.init') + + if('dbms' in kwargs): + self.dbms = kwargs['dbms'] + + if('dbserver' in kwargs): + self.dbserver = kwargs['dbserver'] + + if('userid' in kwargs): + self.userid = kwargs['userid'] + + if('password' in kwargs): + self.password = kwargs['password'] + + if('outpath' in kwargs): + self.outpath = kwargs['outpath'] + + if('db' in kwargs): + self.db = kwargs['db'] + + if('tap_schema' in kwargs): + self.tap_schema = kwargs['tap_schema'] + + # + # Get keyword parameters + # + + if ((self.dbms is None) or (len(self.dbms) == 0)): + self.msg = 'Database name missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + + raise Exception (self.msg) + + if self.debug: + logging.debug('') + logging.debug(f'dbms= {self.dbms:s}') + + + if (self.dbms.lower() == 'oracle'): + + if ((self.dbserver is None) or (len(self.dbserver) == 0)): + self.msg = 'Database server name missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'dbserver= {self.dbserver:s}') + + if ((self.userid is None) or (len(self.userid) == 0)): + self.msg = 'Database userid missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'userid= {self.userid:s}') + + if ((self.password is None) or (len(self.password) == 0)): + self.msg = 'Database password missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'password= {self.password:s}') + + elif (self.dbms.lower() == 'sqlite3'): + + if ((self.db is None) or (len(self.db) == 0)): + self.msg = 'Database name missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'db= {self.db:s}') + + if ((self.tap_schema is None) or (len(self.tap_schema) == 0)): + + self.msg = 'Database table name missing.' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'tap_schema= {self.tap_schema:s}') + + + if ((len(self.outpath) == 0) or (self.outpath is None)): + self.msg = 'Required output path is missing.' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'outpath= {self.outpath:s}') + + # + # { Connect to DBMS + # + if(self.dbms.lower() == 'oracle'): + + import cx_Oracle + try: + self.conn = cx_Oracle.connect(self.userid, + self.password, + self.dbserver) + + if self.debug: + logging.debug('') + logging.debug('connected to Oracle, DB ' + self.dbserver) + + except Exception as e: + + self.status = 'error' + self.msg = 'Failed to connect to cx_Oracle' + + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + + elif (self.dbms.lower() == 'sqlite3'): + + import sqlite3 + + self.db = '' + if('db' in self.connectInfo): + self.db = self.connectInfo['db'] + + if(len(self.db) == 0): + self.msg = 'Failed to retrieve required input parameter'\ + ' [db]' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + + self.tap_schema = '' + if('tap_schema' in self.connectInfo): + self.tap_schema = self.connectInfo['tap_schema'] + + if(len(self.tap_schema) == 0): + self.msg = 'Failed to retrieve required input parameter'\ + ' [tap_schema]' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + if self.debug: + logging.debug('') + logging.debug(f'db= {self.db:s}') + logging.debug(f'tap_schema= {self.tap_schema:s}') + + try: + self.conn = sqlite3.connect(self.db) + + if self.debug: + logging.debug('') + logging.debug('connected to SQLite3, database ' + self.db) + + cmd = 'ATTACH DATABASE ? AS TAP_SCHEMA' + + dbspec = (self.tap_schema,) + + if self.debug: + logging.debug('') + logging.debug('cmd: ' + cmd + '(' + self.tap_schema + ')') + + cursor = self.conn.cursor() + + cursor.execute(cmd, dbspec) + + if self.debug: + logging.debug('') + logging.debug('TAP_SCHEMA attached') + + except Exception as e: + + self.status = 'error' + self.msg = 'Failed to connect to SQLite3 databases' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + else: + self.msg = f'Invalid DBMS: {self.dbms:s}' + self.status = 'error' + #self.__printVosiError__ (self.msg) + raise Exception (self.msg) + + + # + # } end Connect to DBMS + # + + # + # start loop for constructing output: + # + # -- vosi header: tableset tag, + # -- query for schemas: schema tag, + # -- for each schema, query tables: table tag, + # -- for each table, query columns: column tag, + # -- end column tag, + # -- for each table, query foreignkey: foreignkey tag, + # -- end foreignkey tag + # -- end table tag, + # -- end schema tag + # -- end tableset tag + + # + # open ouput file in workdir for vosi table output + # + #vositblpath = self.workdir + '/vositable.xml' + vositblpath = self.outpath + + if self.debug: + logging.debug('') + logging.debug(f'vositblpath = {vositblpath:s}') + + + fp = None + try: + fp = open (vositblpath, 'w') + os.chmod(vositblpath, 0o664) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'open voditblpath exception: {str(e):s}') + + self.msg = 'Failed to open/create vositbl file.' + #self.__printVosiError__(self.msg) + raise Exception (self.msg) + + + # + # write vositable header + # + fp.write ("\n") + fp.write(" \n") + fp.flush() + + # + # Submit database query for TAP_SCHEMA.schemas + # + + self.sql = "select schema_name, description from TAP_SCHEMA.schemas order by schema_index" + + if self.debug: + logging.debug('') + logging.debug(f'sql = {self.sql:s}') + logging.debug('call execute sql') + + cursor = self.conn.cursor() + + try: + self.__executeSql__(cursor, self.sql) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'executeSql exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned executeSql') + + + try: + if self.debug: + logging.debug('') + logging.debug('call getSchemaQueryArr__:') + + self.__getSchemaQueryArr__ (cursor, \ + debug=self.debug) + + if self.debug: + logging.debug('') + logging.debug('returned getSchemaDesc') + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'writeResult exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + self.nschema = len (self.schema_namearr) + + if self.debug: + logging.debug('') + logging.debug(f'nschema= {self.nschema:d}') + logging.debug('') + logging.debug('schema name and desc') + + for i in range (0,self.nschema): + logging.debug(f'{i:d} schema_name={self.schema_namearr[i]:s}') + logging.debug(f'schema_description={self.schema_descarr[i]:s}') + + for ischema in range (0, self.nschema): + # + # { ischema loop + # + # for each schema: + # 1. write header block for each schema, + # + fp.write(" \n") + fp.write(" " + \ + self.schema_namearr[ischema] + "\n") + fp.write(" " + \ + self.schema_descarr[ischema] + "\n") + fp.flush() + + + # + # 2. execute table sql + # + self.sql = "select table_name, description, table_type " + \ + "from TAP_SCHEMA.tables where schema_name='" + \ + self.schema_namearr[ischema] + "' order by table_index" + + if self.debug: + logging.debug('') + logging.debug(f'ischema= {ischema:d}') + logging.debug(f'schema_name= {self.schema_namearr[i]:s}') + logging.debug('') + logging.debug(f'sql = {self.sql:s}') + logging.debug('call execute sql') + + cursor = self.conn.cursor() + + try: + self.__executeSql__(cursor, self.sql) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'executeSql exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned executeSql') + logging.debug('') + logging.debug('call getTableQueryArr') + + # + # 3. getSqlresult of table sql + # re-initialize arrays + # + + table_namearr = [] + table_descarr = [] + table_typearr = [] + + ntable = 0 + try: + ntable = self.__getTableQueryArr__(cursor, table_namearr, \ + table_descarr, table_typearr, debug=1) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'getTableQueryArr exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned getTableQueryArr') + logging.debug(f'ntable= {ntable:d}') + + + for itable in range (0, ntable): + logging.debug ('') + logging.debug (f'itable= {itable:d}') + logging.debug ( + f'table_name= {table_namearr[itable]:s}') + logging.debug ( + f'table_desc= {table_descarr[itable]:s}') + logging.debug ( + f'table_type= {table_typearr[itable]:s}') + + # + # for each table: + # + + for itable in range (0, ntable): + # + # { itable loop + # + # 1. write header block for each table, + # + fp.write(" \n") + fp.write(" " + \ + table_namearr[itable] + "\n") + fp.write(" " + \ + table_descarr[itable] + "\n") + + + # + # 2. execute column sql + # + self.sql = "select column_name, description, unit, ucd, " + \ + "datatype, principal, indexed from TAP_SCHEMA.columns " + \ + "where table_name='" + table_namearr[itable] + \ + "' order by column_index" + + if self.debug: + logging.debug('') + logging.debug(f'sql = {self.sql:s}') + logging.debug('call execute sql') + + cursor = self.conn.cursor() + + try: + self.__executeSql__(cursor, self.sql) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'executeSql exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned executeSql') + + # + # 3. getSqlresult of column sql and write column result + # + try: + self.__writeOneTableResult__ (cursor, fp, **kwargs) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug( + f'writeOneTableResult exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned writeOneTableResult') + + # + # 4. execute foreignkey sq + # 5. getSqlresult of foreignkey sql + # -- write foreignkey result + + self.sql = "select key_id as foreignKey, " + \ + "target_table as targetTable, " + \ + "from_column as fromColumn, " + \ + "target_column as targetColumn, description " + \ + "from TAP_SCHEMA.keys " + \ + "NATURAL JOIN TAP_SCHEMA.key_columns " + \ + "where from_table='" + table_namearr[itable] + "'" + + if self.debug: + logging.debug('') + logging.debug(f'foreign key sql:'); + logging.debug(f'sql= {self.sql:s}'); + logging.debug('call execute sql') + + cursor = self.conn.cursor() + + try: + self.__executeSql__(cursor, self.sql) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug(f'executeSql exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + if self.debug: + logging.debug('') + logging.debug('returned executeSql') + + # + # 5. getSqlresult of foreignkey sql and write foreignkey result + # + try: + self.__writeForeignKey__ (cursor, fp, **kwargs) + + except Exception as e: + + if self.debug: + logging.debug('') + logging.debug( + f'writeForeignKey exception: {str(e):s}') + + #self.__printVosiError__ (str(e)) + raise Exception (str(e)) + + + + if self.debug: + logging.debug('') + logging.debug('returned writeForeignKey') + + + # + # 6. write end block for each table + # + fp.write("
\n") + fp.flush() + + # + # } end itable loop + # + + # + # 2. write end block for each schema + # + fp.write(" \n") + fp.flush() + + # + # } end ischema loop and write end block for xml file + # + fp.write("\n") + fp.flush() + + # + # } end vosiTables.init + # + + + def __getTableQueryArr__ (self, cursor, table_namearr, table_descarr, \ + table_typearr, **kwargs): + # + # { vosiTable.getTableQueryArr + # + # this query contains only three char columns: + # table_name, description, and table_type; + # + + if self.debug: + logging.debug('') + logging.debug('Enter getTableQueryArr') + + size = 0 + + table_colname = [] + table_dbtype = [] + + nfetch = self.arraysize + cursor.arraysize = nfetch + + if self.debug: + logging.debug('') + logging.debug(f'nfetch = {nfetch:d}') + + + i = 0 + for col in cursor.description: + + # + # { For loop for cursor description: + # cursor description is a list with ncols tuples: + # each tuple + # contains each output colname's name, datatype, + # dbdatatype, size, etc... + # + + if self.debug: + logging.debug('') + logging.debug('get cursor description:') + logging.debug(f'i = {i:d} col = ' + str(col)) + + # + # Extract colname, dbdatatype, size, precision, scale from + # col array + # + colname = str(col[0]).lower() + table_colname.append(colname) + + if self.debug: + logging.debug('') + logging.debug(f'colname(lower) = {colname:s}') + + + dbtype = '' + size = None + + dbdatatypestr = str(col[1]) + + if self.debug: + logging.debug('') + logging.debug(f'dbdatatypestr = {dbdatatypestr:s}') + + ind = dbdatatypestr.find("VARCHAR") + if(ind != -1): + dbtype = 'STRING' + + ind = dbdatatypestr.find("STRING") + if(ind != -1): + dbtype = 'STRING' + + ind = dbdatatypestr.find("NUMBER") + if(ind != -1): + dbtype = 'NUMBER' + + if self.debug: + logging.debug(f'dbtype = {dbtype:s}') + + table_dbtype.append(dbtype) + + i = i + 1 + + # + # } end of for loop for analysing cursor description + # + + # + # Start fetching data lines + # + + if self.debug: + logging.debug('') + logging.debug('Start fetching data:') + + if self.debug: + logging.debug('') + logging.debug(f'nfetch= {nfetch:d}') + + irow = 0 + ntot = 0 + + while True: + + # + # { start of while loop for fetching data lines; + # max 10000 lines at a time + # + + rows = cursor.fetchmany() + + nrec = len(rows) + + if self.debug: + logging.debug(f'nrec = {nrec:d}') + logging.debug('') + + + for ll in range(0, nrec): + + # + # { Beginning ll loop: one row + # + + row = rows[ll] + + if self.debug: + logging.debug ('') + logging.debug (f'len(row)= {len(row):d}') + logging.debug ('') + logging.debug ('row:') + logging.debug(row) + + for i in range(0, len(row)): + + # + # { Beginning i loop: one col + # + data = '' + + if (table_dbtype[i] != 'STRING'): + + data = str(row[i]) + else: + data = row[i] + + if self.debug: + logging.debug (f'i= {i:d}, data= {data:s}') + logging.debug (f'table_colname= {table_colname[i]:s}') + + if (table_colname[i].lower() == 'table_name'): + table_namearr.append (data) + + elif (table_colname[i].lower() == 'description'): + table_descarr.append (data) + + elif (table_colname[i].lower() == 'table_type'): + table_typearr.append (data) + + # + # } end of i loop + # + + irow = irow + 1 + + # + # } end of ll loop + # + if self.debug: + logging.debug('') + logging.debug(table_namearr) + logging.debug(table_descarr) + logging.debug(table_typearr) + logging.debug('----------------------------------------') + + + ntot = ntot + nrec + + if(len(rows) < cursor.arraysize): + break + + if self.debug: + logging.debug ('') + logging.debug (f'ntot= {ntot:d}') + # + # } end while loop for fetching data lines + # + + return (ntot) + + # + # } end vosiTables.getTableQueryArr + # + + + + def __getSchemaQueryArr__ (self, cursor, **kwargs): + # + # { vosiTables.getSchemaQueryArr + # + # this query contains only two columns: + # schema_name and schema_description + # + if self.debug: + logging.debug('') + logging.debug(f'Enter __getSchemaQueryArr__\n') + + size = 0 + nfetch = self.arraysize + cursor.arraysize = nfetch + + if self.debug: + logging.debug('') + logging.debug(f'nfetch= {nfetch:d}') + + + i = 0 + for col in cursor.description: + + # + # { For loop for cursor description: + # cursor description is a list with ncols tuples: + # each tuple # contains each output colname's name, datatype, + # dbdatatype, size, etc... + # + + if self.debug: + logging.debug('') + logging.debug('----------------------------------------------') + logging.debug(f'i = {i:d} col = ' + str(col)) + + # + # Extract colname, dbdatatype, size, precision, scale from + # col array + # + + colname = str(col[0]).lower() + + if self.debug: + logging.debug('') + logging.debug(f'colname(lower) = {colname:s}') + + # + # { Extract dbdatatype, display_size, precision, scale from + # col_array: + # col[1], col[2], col[4], and col[5] + # + + dbdatatype = '' + size = None + precision = None + scale = None + + dbdatatypestr = str(col[1]) + + if self.debug: + logging.debug('') + logging.debug(f'dbdatatypestr = {dbdatatypestr:s}') + + ind = dbdatatypestr.find("VARCHAR") + if(ind != -1): + dbdatatype = 'VARCHAR' + + ind = dbdatatypestr.find("STRING") + if(ind != -1): + dbdatatype = 'STRING' + + size = None + if(col[2] is not None): + + size = col[2] + if(len(colname) > size): + size = len(colname) + + if self.debug: + logging.debug(f'size = {size:d}') + + # + # } end extract dbdatatype, display_size, precision, + # scale from + # col_array: col[1], col[2], col[4], and col[5] + + + dbtype = '' + datatype = '' + fmt = '' + width = 0 + + if(dbdatatype == 'STRING') or (dbdatatype == 'VARCHAR'): + + # + # { dbdatatype == string + # + + width = 0 + if(size is not None): + width = size + + if(len(colname) > width): + width = len(colname) + + datatype = 'char' + dbtype = dbdatatype + fmt = str(width) + 's' + + # + # } end dbdatatype == STRING + # + + else: + # + # { unknown type: default to char + # + + datatype = 'char' + dbtype = dbdatatype + + width = 200 + if(len(colname) > width): + width = len(colname) + fmt = str(width) + 's' + + # + # } end unknown type + # + + if self.debug: + logging.debug('') + logging.debug(f' width = {width:d}') + logging.debug(f' dbtype = {dbtype:s}') + logging.debug(f' datatype = {datatype:s}') + logging.debug(f' fmt = {fmt:s}') + + i = i + 1 + + # + # } end of for loop for analysing cursor description + # + + if self.debug: + logging.debug('') + logging.debug('Start fetching data') + logging.debug('----------------------------------------') + + # + # Start fetching data lines + # + + nfetch = self.arraysize + + cursor.arraysize = nfetch + + irow = 0 + ntot = 0 + + while True: + + # + # { start of while loop for fetching data lines; + # max 10000 lines at a time + # + + rows = cursor.fetchmany() + + nrec = len(rows) + + if self.debug: + logging.debug(f'nrec = {nrec:d}') + logging.debug('') + + + for ll in range(0, nrec): + + # + # { Beginning ll loop: one row + # + + row = rows[ll] + + if self.debug: + logging.debug(row) + logging.debug(f'row[0]= {row[0]:s}') + logging.debug(f'row[1]= {row[1]:s}') + + + self.schema_namearr.append(row[0]) + self.schema_descarr.append(row[1]) + + if self.debug: + logging.debug(row) + logging.debug(self.schema_namearr) + logging.debug(self.schema_descarr) + + irow = irow + 1 + + # + # } end of ll loop + # + + if self.debug: + logging.debug('----------------------------------------') + logging.debug(f'irow= {irow:d}') + + ntot = ntot + nrec + + if(len(rows) < cursor.arraysize): + break + + # + # } end while loop for fetching data lines + # + + return + + # + # } end vosiTables.getSchemaQueryArr + # + + + def __writeOneTableResult__ (self, cursor, fp, **kwargs): + + # + # {vosiTables.writeOneTableResult + # + if self.debug: + logging.debug('') + logging.debug('Enter __writeOneTableResult') + + cursor.arraysize = self.arraysize + + dbtype = [] + + i = 0 + for col in cursor.description: + + # + # { For loop for cursor description: + # cursor description is a list with ncols tuples: + # each tuple + # contains each output colname's name, datatype, + # dbdatatype, size, etc... + # + + if self.debug: + logging.debug('') + logging.debug(f'i = {i:d} col = ' + str(col)) + + # + # Extract colname, dbdatatype, size, precision, scale from + # col array + # + + colname = str(col[0]).lower() + + if self.debug: + logging.debug('') + logging.debug(f'colname(lower) = {colname:s}') + + # + # { Extract dbdatatype, display_size, precision, scale from + # col_array: + # col[1], col[2], col[4], and col[5] + # + + dbdatatype = '' + size = None + + dbdatatypestr = str(col[1]) + + if self.debug: + logging.debug('') + logging.debug(f'dbdatatypestr = {dbdatatypestr:s}') + + ind = dbdatatypestr.find("VARCHAR") + if(ind != -1): + dbdatatype = 'STRING' + + ind = dbdatatypestr.find("STRING") + if(ind != -1): + dbdatatype = 'STRING' + + if self.debug: + logging.debug(f'dbdatatype = {dbdatatype:s}') + + dbtype.append(dbdatatype) + # + # } end extract dbdatatype from col_array: col[1] + # + + i = i + 1 + + # + # } end of for loop for analysing dd's cursor description + # + + # + # Start fetching data lines + # + + nfetch = self.arraysize + cursor.arraysize = nfetch + + irow = 0 + ntot = 0 + + while True: + + # + # { start of while loop for fetching data lines; + # max 10000 lines at a time + # + + rows = cursor.fetchmany() + nrec = len(rows) + + if self.debug: + logging.debug(f'nrec = {nrec:d}') + logging.debug('') + + for ll in range(0, nrec): + + + # + # { Beginning ll loop: one row + # + + row = rows[ll] + + col_name = '' + col_desc = '' + col_unit = '' + col_ucd = '' + col_datatype = '' + col_principal = '' + col_indexed = '' + + data = '' + + for i in range(0, len(row)): + + if (row[i] is None): + data = 'None' + elif (dbtype[i] == 'STRING'): + data = row[i] + else: + data = str(row[i]) + + if self.debug: + logging.debug('') + logging.debug(f'i= {i:d}, data= {data:s}') + + if (i == 0): + col_name = data + elif (i == 1): + col_desc = data + elif (i == 2): + col_unit = data + elif (i == 3): + col_ucd = data + elif (i == 4): + col_datatype = data + elif (i == 5): + col_principal = data + elif (i == 6): + col_indexed = data + + irow = irow + 1 + + if col_datatype == 'timestamp': + col_datatype = 'char' + + + # + # for row, write column block + # + fp.write(" \n") + fp.write(" " + col_name + "\n") + fp.write(" " + \ + "\n") + + if (col_unit != 'None'): + fp.write(" " + col_unit + "\n") + + if (col_ucd != 'None'): + fp.write (" " + col_ucd + "\n") + + fp.write (" " + \ + "" + \ + col_datatype + "\n") + + if (col_principal != 'None' and col_principal != 0 \ + and col_principal != '0'): + fp.write(" " + \ + "principal\n") + + if (col_indexed != 'None' and col_indexed != 0 \ + and col_indexed != '0'): + fp.write(" " + "indexed\n") + + fp.write(" \n") + + + # + # } end of ll loop + # + + ntot = ntot + nrec + + if(len(rows) < cursor.arraysize): + break + + if self.debug: + logging.debug('') + logging.debug(f'ntot= {ntot:d}') + + # + # } end of while loop for fetching data lines + # + + return + + # + # }end vosiTables.writeOneTableResult + # + + + def __writeForeignKey__ (self, cursor, fp, **kwargs): + + # + # {vosiTables.writeForeignKey + # + if self.debug: + logging.debug('') + logging.debug('Enter __writeForeignKey') + + cursor.arraysize = self.arraysize + + dbtype = [] + + i = 0 + for col in cursor.description: + + # + # { For loop for cursor description: + # cursor description is a list with ncols tuples: + # each tuple + # contains each output colname's name, datatype, + # dbdatatype, size, etc... + # + + if self.debug: + logging.debug('') + logging.debug(f'i = {i:d} col = ' + str(col)) + + # + # Extract colname, dbdatatype, size, precision, scale from + # col array + # + + colname = str(col[0]).lower() + + if self.debug: + logging.debug('') + logging.debug(f'colname(lower) = {colname:s}') + + # + # { Extract dbdatatype, display_size, precision, scale from + # col_array: + # col[1], col[2], col[4], and col[5] + # + + dbdatatype = '' + size = None + + dbdatatypestr = str(col[1]) + + if self.debug: + logging.debug('') + logging.debug(f'dbdatatypestr = {dbdatatypestr:s}') + + ind = dbdatatypestr.find("VARCHAR") + if(ind != -1): + dbdatatype = 'STRING' + + ind = dbdatatypestr.find("STRING") + if(ind != -1): + dbdatatype = 'STRING' + + if self.debug: + logging.debug(f'dbdatatype = {dbdatatype:s}') + + dbtype.append(dbdatatype) + # + # } end extract dbdatatype from col_array: col[1] + # + + i = i + 1 + + # + # } end of for loop for analysing dd's cursor description + # + + # + # Start fetching data lines + # + + nfetch = self.arraysize + cursor.arraysize = nfetch + + irow = 0 + ntot = 0 + + while True: + + # + # { start of while loop for fetching data lines; + # max 10000 lines at a time + # + + rows = cursor.fetchmany() + nrec = len(rows) + + if self.debug: + logging.debug(f'nrec = {nrec:d}') + logging.debug('') + + for ll in range(0, nrec): + + + # + # { Beginning ll loop: one row + # + + row = rows[ll] + + keyid = '' + targettbl = '' + fromcol = '' + targetcol= '' + desc = '' + + data = '' + for i in range(0, len(row)): + + if (row[i] is None): + data = 'None' + elif (dbtype[i] == 'STRING'): + data = row[i] + else: + data = str(row[i]) + + if self.debug: + logging.debug('') + logging.debug(f'i= {i:d}, data= {data:s}') + + if (i == 0): + keyid = data + elif (i == 1): + targettbl = data + elif (i == 2): + fromcol = data + elif (i == 3): + targetcol = data + elif (i == 4): + desc = data + + irow = irow + 1 + + # + # for row, write column block + # + fp.write(" \n") + fp.write(" " + targettbl + \ + "\n") + + fp.write(" \n") + fp.write(" " + fromcol + \ + "\n") + fp.write(" " + targetcol + \ + "\n") + fp.write(" \n") + + fp.write(" " + \ + desc + "\n") + fp.write(" \n") + + # + # } end of ll loop + # + + ntot = ntot + nrec + + if(len(rows) < cursor.arraysize): + break + + if self.debug: + logging.debug('') + logging.debug(f'ntot= {ntot:d}') + + # + # } end of while loop for fetching data lines + # + + return + + # + # }end vosiTables.writeOneTableResult + # + + + + + + + + + + + def __executeSql__(self, cursor, sql, **kwargs): + # + # { vosiTables.executeSql + # + + debug = 0 + + if('debug' in kwargs): + debug = kwargs['debug'] + + if self.debug: + logging.debug('') + logging.debug('Enter executeSql') + logging.debug(f'sql:= {sql:s}') + + try: + cursor.execute(sql) + except Exception as e: + + msg = str(e).replace('"', "'") + + if self.debug: + logging.debug('') + logging.debug(f'create table exception: {str(msg):s}') + + # raise Exception(str(e)) + raise Exception(msg) + + # + # } end vosiTables.executeSql + # + + + def __printVosiError__(self, errmsg): + + # + # { + # + + httphdr = "HTTP/1.1 500 ERROR\r" + + #print("HTTP/1.1 200 OK\r") + + print(httphdr) + + print("Content-type: text/xml\r") + print("\r") + + print('') + print('') + print('') + print('') + + print(errmsg) + + print('') + print('') + print('') + + sys.stdout.flush() + sys.exit() + + # + # } end of printError + # + + +# +# } end class vosiTables: +# + +def main(): + + print ('Enter main') + + sys.tracebacklimit = 0 + + parser = argparse.ArgumentParser(description='vosiTables') + + parser.add_argument('--dbms', required=True, + help='database (e.g. oracle, sqlite, etc..) to use.') + + parser.add_argument('--dbserver', required=False, + help='Oracle database server (dbserver) to use.') + + parser.add_argument('--userid', required=False, + help='Oracle dbserver userid (userid) to use.') + + parser.add_argument('--password', required=False, + help='Oracle dbserver (password) to use.') + + parser.add_argument('--db', required=False, + help='Sqlite database name to use.') + + parser.add_argument('--tap_schema', required=False, + help='Sqlite database table name to use.') + + parser.add_argument('--outpath', required=True, + help='outpath for the result.') + + parser.add_argument('--debug', required=False, + help='debugflag: 1/0.') + + args = parser.parse_args() + + dbms = '' + dbserver = '' + userid = '' + password = '' + outpath = '' + debug = 0 + + db = '' + tap_schema = '' + + if (args.dbms is not None): + dbms = args.dbms + print (f'dbms= {dbms:s}') + + if (args.dbserver is not None): + dbserver = args.dbserver + print (f'dbserver= {dbserver:s}') + + if (args.userid is not None): + userid = args.userid + print (f'userid= {userid:s}') + + + if (args.password is not None): + password = args.password + print (f'password= {password:s}') + + + if (args.outpath is not None): + outpath = args.outpath + print (f'outpath= {outpath:s}') + + + if (args.debug is not None): + debug = args.debug + print (f'debug= {debug:s}') + + if (args.db is not None): + db = args.db + print (f'db= {db:s}') + + if (args.tap_schema is not None): + tap_schema = args.tap_schema + print (f'tap_schema= {tap_schema:s}') + + + if debug: + debugfname = './vosi.debug' + + logging.basicConfig(filename=debugfname, \ + level=logging.DEBUG) + + print (f'debug turned on') + + logging.debug('') + logging.debug('debug turned on') + logging.debug(f'dbms= {dbms:s}') + logging.debug(f'dbserver= {dbserver:s}') + logging.debug(f'userid= {userid:s}') + logging.debug(f'password= {password:s}') + logging.debug(f'outpath= {outpath:s}') + logging.debug(f'debug= {debug:s}') + + + vositbl = None + try: + if debug: + logging.debug('') + logging.debug('call vosiTables') + + vositbl = vosiTables (dbms=dbms, \ + dbserver=dbserver, \ + userid=userid, \ + password=password, \ + outpath=outpath, \ + debug=debug) + + if debug: + logging.debug('') + logging.debug('vosiTables initialized') + + print('vosiTables initialized') + + except Exception as e: + + if debug: + logging.debug('') + logging.debug(f'vosiTable.init failed: {str(e):s}') + + print(f'vosiTable.init failed: {str(e):s}') + + +if __name__ == "__main__": + main() diff --git a/TAP/configparam.py b/TAP/configparam.py index 58701e8..891dfa9 100644 --- a/TAP/configparam.py +++ b/TAP/configparam.py @@ -9,6 +9,8 @@ import configobj +from TAP import compat + class configParam: @@ -71,6 +73,7 @@ def __init__(self, path, **kwargs): raise Exception(self.msg) confobj = configobj.ConfigObj(self.configpath) + confobj, self.compat = compat.load_config(confobj) if self.debug: logging.debug('') @@ -647,6 +650,7 @@ def __init__(self, path, **kwargs): self.connectInfo['columns_table'] = self.columns_table self.connectInfo['keys_table'] = self.keys_table self.connectInfo['key_columns_table'] = self.key_columns_table + self.connectInfo['compat'] = self.compat self.connectInfo['port'] = self.dbport self.connectInfo['socket'] = self.socket self.connectInfo['dbschema'] = self.dbschema diff --git a/TAP/datadictionary.py b/TAP/datadictionary.py index 7d27719..a11cc2d 100644 --- a/TAP/datadictionary.py +++ b/TAP/datadictionary.py @@ -186,7 +186,8 @@ def __init__(self, conn, table, connectInfo, **kwargs): # For SQLite the schema is ATTACHed under a name held in # `tap_schema_file`; `tap_schema` is the file path. For # Oracle/Postgres/MySQL `tap_schema` is the schema name itself. - schema_name = self.connectInfo.get("tap_schema_file") \ + schema_name = (self.connectInfo.get("tap_schema_file") + if self.connectInfo.get("dbms") == "sqlite3" else None) \ or self.connectInfo["tap_schema"] sql = "select * from " + schema_name + "." + self.connectInfo["columns_table"] + " where lower(table_name) = " \ diff --git a/TAP/tablevalidator.py b/TAP/tablevalidator.py index 74236b7..fa80940 100644 --- a/TAP/tablevalidator.py +++ b/TAP/tablevalidator.py @@ -37,8 +37,9 @@ def __init__(self, conn, connectInfo=None, debug=0): # there's no ATTACH and `tap_schema` is the schema name itself. # For SQLite, `tap_schema` is the FILE PATH and using it directly # in a `SELECT ... FROM .tables` reference produces a SQL - # syntax error. Prefer the more-specific key when it's set. - if 'tap_schema_file' in connectInfo: + # syntax error. So `tap_schema_file` is used only for SQLite + # (dbms == 'sqlite3'); other databases use `tap_schema`. + if connectInfo.get('dbms') == 'sqlite3' and 'tap_schema_file' in connectInfo: self.tap_schema = connectInfo['tap_schema_file'] elif 'tap_schema' in connectInfo: self.tap_schema = connectInfo['tap_schema'] diff --git a/TAP/tap.py b/TAP/tap.py index 3442d23..8ca2f97 100755 --- a/TAP/tap.py +++ b/TAP/tap.py @@ -19,12 +19,12 @@ from bs4 import BeautifulSoup from spatial_index import SpatialIndex +from TAP import compat from TAP.configparam import configParam from TAP.propfilter import propFilter from TAP.tablenames import TableNames from TAP.tablevalidator import TableValidationError, TableValidator from TAP.tapquery import tapQuery -from TAP.vositables import vosiTables class Tap: @@ -143,6 +143,7 @@ class Tap: uwsheader = '' + compat = frozenset() # active compat names; set once TAP.conf loads def __init__(self, **kwargs): @@ -470,6 +471,8 @@ def __run__(self, **kwargs): self.config = None try: self.config = configParam(self.configpath, instance=self.instance, debug=self.debug) + self.compat = self.config.compat + compat.warn_once_per_day(self.compat, self.config.workdir) if self.debug: logging.debug('') @@ -870,7 +873,8 @@ def __run__(self, **kwargs): logging.debug('case: getStatus') try: - self.__getStatus__(self.workdir, self.id, self.statuskey, \ + self.__getStatus__(self.workdir, self.id, + compat.uws_url_key(self.compat, self.statuskey), \ self.param) except Exception as e: @@ -977,7 +981,7 @@ def __run__(self, **kwargs): self.statdict['starttime'] = job['uws:startTime'] self.statdict['endtime'] = job['uws:endTime'] self.statdict['destruction'] = job['uws:destruction'] - self.statdict['duration'] = job['uws:executionduration'] + self.statdict['duration'] = compat.duration_value(job) if self.debug: logging.debug ('') @@ -1072,7 +1076,7 @@ def __run__(self, **kwargs): starttime = stime.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-4] destruction = \ - destructtime.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-4] + 'Z' + destructtime.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-4] + compat.destruction_suffix(self.compat) if self.debug: logging.debug('') @@ -1280,24 +1284,42 @@ def __run__(self, **kwargs): # } end async bogus value case # - if self.debug: - logging.debug('') - logging.debug ('call writeStatusMsg') - logging.debug (f'statuspath= {self.statuspath:s}') + if (self.param['phase'] == 'RUN'): + # + # Publish the job, answer the client, and keep running + # the query in a detached child. + # - self.__writeStatusMsg__(self.statuspath, self.statdict, - self.param) + self.__respondAsyncAndDetach__() - if self.debug: - logging.debug('') - logging.debug ('call printAsyncResponse') + if self.debug: + logging.debug('') + logging.debug('async response sent; job running detached') - self.__printAsyncResponse__(self.statusurl) + else: - if self.debug: - logging.debug('') - logging.debug('returned printAsyncResponse') + # + # ABORT and unrecognized phases are terminal: the branch + # above has already decided the phase, so write it, + # answer the client and stop. Falling through to the + # query path below would run the job the client just + # aborted and overwrite ABORTED (or ERROR) with + # COMPLETED. + # + + if self.debug: + logging.debug('') + logging.debug ('terminal async phase= ' + f"{self.statdict['phase']:s}") + logging.debug (f'statuspath= {self.statuspath:s}') + + self.__writeStatusMsg__(self.statuspath, self.statdict, + self.param) + + self.__printAsyncResponse__(self.statusurl) + + sys.exit() # # } end async submit case @@ -1482,7 +1504,7 @@ def __run__(self, **kwargs): try: TableValidator.validate_statement(query_adql, debug=self.debug) except Exception as e: - errcode = '403' if isinstance(e, TableValidationError) else '400' + errcode = compat.table_denied_status(self.compat) if isinstance(e, TableValidationError) else '400' if(self.tapcontext == 'async'): self.phase = 'ERROR' self.__writeAsyncError__(str(e), self.statuspath, @@ -1724,7 +1746,7 @@ def __run__(self, **kwargs): # TableValidationError means the query referenced a table # not in TAP_SCHEMA — access denied (403). All other errors # are bad requests (400). - errcode = '403' if isinstance(e, TableValidationError) else '400' + errcode = compat.table_denied_status(self.compat) if isinstance(e, TableValidationError) else '400' if(self.tapcontext == 'async'): @@ -1785,7 +1807,7 @@ def __run__(self, **kwargs): # TableValidationError means the query referenced a table # not in TAP_SCHEMA — access denied (403). All other errors # are bad requests (400). - errcode = '403' if isinstance(e, TableValidationError) else '400' + errcode = compat.table_denied_status(self.compat) if isinstance(e, TableValidationError) else '400' if(self.tapcontext == 'async'): @@ -2046,7 +2068,7 @@ def __printStatus__(self, key, retval, outtype, **kwargs): if(outtype == 'xml'): - print("Content-type: text/xml\r") + print("Content-type: %s\r" % compat.uws_content_type(self.compat)) print("\r") print('') @@ -2148,7 +2170,7 @@ def __getStatus__(self, workdir, workspace, key, param, **kwargs): logging.debug('Return status.xml to user and exit.') print("HTTP/1.1 200 OK\r") - print("Content-type: text/xml\r") + print("Content-type: %s\r" % compat.uws_content_type(self.compat)) print("\r") print(data) sys.exit() @@ -2246,7 +2268,7 @@ def __getStatus__(self, workdir, workspace, key, param, **kwargs): # { Single value return # retval = 'None' - keystr = 'uws:' + key + keystr = 'uws:' + compat.uws_key(job, key) outstr = '' if((key == 'phase') @@ -2470,7 +2492,7 @@ def __getStatus__(self, workdir, workspace, key, param, **kwargs): if(format == 'json'): print("Content-type: application/json\r") elif(format == 'votable'): - print("Content-type: text/xml\r") + print("Content-type: %s\r" % compat.votable_content_type(self.compat)) else: print("Content-type: text/plain\r") print("\r") @@ -2535,6 +2557,12 @@ def __printError__(self, fmt, errmsg, **kwargs): if ('errcode' in kwargs): errcode = kwargs['errcode'] + nea = compat.error_document(self.compat, errmsg, errcode) + if nea is not None: # nea-errors replaces the whole response + sys.stdout.write(nea) + sys.stdout.flush() + sys.exit() + httphdr = "HTTP/1.1 " + str(errcode) + " ERROR\r" print(httphdr) @@ -2663,7 +2691,7 @@ def __printSyncResult__(self, resultpath, format, **kwargs): if(format == 'json'): print("Content-type: application/json\r") elif(format == 'votable'): - print("Content-type: text/xml\r") + print("Content-type: %s\r" % compat.votable_content_type(self.compat)) else: print("Content-type: text/plain\r") print("\r") @@ -2726,25 +2754,210 @@ def __printSyncResponse__(self, status, msg, resulturl, format, **kwargs): def __printAsyncResponse__(self, statusurl, **kwargs): # - # async: return statusurl and kill the parent process + # async: point the client at the job's status URL. + # + # The body is one line, but it still needs a Content-Length. This + # is an nph- script, so nothing downstream supplies one, and a + # reverse proxy in front of the CGI has no other way to tell + # where the response ends. Earlier versions omitted it and let + # the connection dying stand in for the end of the message. # - print("HTTP/1.1 303 See Other\r") - print("Location: %s\r\n\r" % statusurl) - print("Redirect Location: %s" % statusurl) + body = 'Redirect Location: %s\n' % statusurl + + sys.stdout.write('HTTP/1.1 303 See Other\r\n') + sys.stdout.write('Location: %s\r\n' % statusurl) + sys.stdout.write(compat.async_submit_headers(self.compat, body)) + sys.stdout.write('\r\n') + sys.stdout.write(body) sys.stdout.flush() - time.sleep(2.0) + if self.debug: + logging.debug('') + logging.debug(f'async response sent: statusurl= {statusurl:s}') + + return + + def __respondAsyncAndDetach__(self, **kwargs): + + # + # { An async submit has to finish an HTTP response now and keep + # executing the query afterwards, and under CGI those two pull + # against each other: the web server completes the response + # when the script closes stdout, and mod_cgi terminates the + # script once the request is cleaned up. The process that runs + # the query can therefore be neither the one holding stdout nor + # part of the request. # - # Shut down parent program + # So fork. The parent names the child as the job's runId, + # writes the status document, sends the 303 and exits, which + # ends the response the way any other CGI would. The child + # leaves the request's process group, points its standard + # streams away from the server pipe, waits for the parent to + # confirm the job is published, and returns to run the query. # + # Earlier versions sent the response and then SIGKILLed + # os.getppid(). Under CGI that parent is the web server child + # serving the request: killing it truncates the response, and + # behind a reverse proxy it leaves the proxy holding a dead + # upstream connection, which the proxy reports as 503 on the + # next request or two routed over it. + # + + try: + readfd, writefd = os.pipe() + pid = os.fork() + + except OSError as e: + + # + # No fork available: answer the request from this process and + # run the job here. The response is still well formed; the + # job now lives and dies with the request. + # + + logging.error(f'Could not fork async worker: {str(e)}') + + self.__writeStatusMsg__(self.statuspath, self.statdict, + self.param) + + self.__printAsyncResponse__(self.statusurl) + + return + + if (pid > 0): + # + # { parent: publish the job, answer the client, exit + # + os.close(readfd) + + self.statdict['process_id'] = pid - os.kill(os.getppid(), signal.SIGKILL) + published = False + + try: + self.__writeStatusMsg__(self.statuspath, self.statdict, + self.param) + + published = True + + self.__printAsyncResponse__(self.statusurl) + + except Exception as e: + + # + # Most likely the client or the proxy hung up while the + # response was going out. Nothing can be sent about it + # now; the job's own fate is decided in the finally + # clause below. + # + + logging.error(f'Async response failed: {str(e)}') + + finally: + + # + # The child blocks on this byte, so a fast query cannot + # overwrite the status document written above. Once the + # job is published the child has to be released even if + # the response itself failed: a client that hung up does + # not make the job go away, and a child that exits here + # would leave the job stuck in EXECUTING forever. If the + # status write is what failed, the byte is withheld and + # the child exits, because there is no job document for + # it to update. + # + + if published: + try: + os.write(writefd, b'1') + + except OSError as e: + logging.error( + f'Could not release async worker: {str(e)}') + + os.close(writefd) + + sys.exit(0) + # + # } end parent + # + + # + # { child: detach from the request, then run the query + # + os.close(writefd) + + self.pid = os.getpid() + self.statdict['process_id'] = self.pid + + self.__detachFromServer__() + + published = b'' + + try: + published = os.read(readfd, 1) + + except OSError as e: + logging.error(f'Async worker handshake failed: {str(e)}') + + os.close(readfd) + + if (len(published) == 0): + + # + # The parent exited before it published the job, so there is + # no status document for this run to update. + # + + logging.error('Async parent exited before publishing the job') + + os._exit(1) if self.debug: logging.debug('') - logging.debug('parent process killed') + logging.debug(f'async worker detached: pid= {self.pid:d}') + + return + # + # } end child + # + + + def __detachFromServer__(self, **kwargs): + + # + # Give up the web server's request context: leave the request's + # process group so the server cannot reap this process along with + # the request, and replace the inherited stdio with /dev/null so + # the response the parent just sent is neither held open by nor + # corrupted from here. + # + + try: + os.setsid() + + except OSError as e: + logging.error(f'setsid failed in async worker: {str(e)}') + + try: + sys.stdout.flush() + sys.stderr.flush() + + except (OSError, ValueError) as e: + logging.error(f'Could not flush async worker stdio: {str(e)}') + + devnull = os.open(os.devnull, os.O_RDWR) + + try: + os.dup2(devnull, 0) + os.dup2(devnull, 1) + os.dup2(devnull, 2) + + finally: + if (devnull > 2): + os.close(devnull) return @@ -2833,7 +3046,8 @@ def __writeStatusMsg__(self, statuspath, statdict, param, **kwargs): fp.write(f" {statdict['endtime']:s}\n") - fp.write(f" {statdict['duration']:d}\n") + el = compat.duration_element(self.compat) + fp.write(f" {statdict['duration']:d}\n") if (statdict['destruction'] is None): fp.write(' \n') @@ -2974,7 +3188,7 @@ def __printVosiTables__ (self, connectInfo, vosipath, **kwargs): logging.debug('') logging.debug('call vosiTables') - vosiTables (connectInfo=self.config.connectInfo, dbms=dbms, \ + compat.vosi_tables_class(self.compat) (connectInfo=self.config.connectInfo, dbms=dbms, \ dbserver=dbserver, \ userid=userid, \ password=password, \ @@ -3037,6 +3251,7 @@ def __printVosiAvailability__ (self, **kwargs): # { printVosiAvailability # + sys.stdout.write(compat.vosi_head(self.compat)) print ('') print ('') print ('TAP service available.') print ('') + sys.stdout.write(compat.vosi_tail(self.compat)) sys.exit() # @@ -3060,6 +3276,7 @@ def __printVosiCapability__ (self, **kwargs): # { printVosiCapability # + sys.stdout.write(compat.vosi_head(self.compat)) print ('') print ('') print ('') + sys.stdout.write(compat.vosi_tail(self.compat)) sys.exit() # diff --git a/TAP/tapquery.py b/TAP/tapquery.py index 47bf0db..76129b4 100644 --- a/TAP/tapquery.py +++ b/TAP/tapquery.py @@ -665,6 +665,7 @@ def __init__(self, **kwargs): maxrec=self.maxrec, arraysize=self.arraysize, coldesc=self.coldesc, + compat=self.connectInfo.get('compat', frozenset()), racol=self.racol, deccol=self.deccol, dbms=self.dbms, \ diff --git a/TAP/vositables.py b/TAP/vositables.py index d414081..2b15001 100644 --- a/TAP/vositables.py +++ b/TAP/vositables.py @@ -211,7 +211,7 @@ def __init__(self, **kwargs): # # { Connect to DBMS # - if('connectInfo' in kwargs): + if('connectInfo' in kwargs): self.connectInfo = kwargs['connectInfo'] else: self.msg = 'Required connectInfo dict is missing.' diff --git a/TAP/writeresult.py b/TAP/writeresult.py index a7a24e1..1d499f1 100644 --- a/TAP/writeresult.py +++ b/TAP/writeresult.py @@ -13,7 +13,7 @@ # import logging -from TAP import writerecs +from TAP import compat, writerecs class writeResult: @@ -153,6 +153,8 @@ def __init__(self, cursor, workdir, dd, **kwargs): if('coldesc' in kwargs): self.coldesc = kwargs['coldesc'] + self.compat = kwargs.get('compat', compat.NONE) + self.arraysize = 10000 if('arraysize' in kwargs): self.arraysize = kwargs['arraysize'] @@ -932,7 +934,7 @@ def __init__(self, cursor, workdir, dd, **kwargs): fmtarr.append(fmt) widtharr.append(width) unitsarr.append(units) - descarr.append(desc) + descarr.append(compat.field_description(self.compat, desc, coltype)) i = i + 1 diff --git a/requirements-test.txt b/requirements-test.txt index 343732f..66043b2 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -10,6 +10,10 @@ configobj sqlparse astropy beautifulsoup4 +# tap.py asks BeautifulSoup for the 'lxml' tree builder by name when it +# reads a job status document (TAP/tap.py:904, 2162), so bs4 alone is not +# enough for the async flow. +lxml xmltodict # pyneid is not currently used — tests/test_pyneid_compat.py is skipped diff --git a/setup.py b/setup.py index ed0d920..b660788 100644 --- a/setup.py +++ b/setup.py @@ -4,7 +4,7 @@ from setuptools import setup setup(name='nexsciTAP', - version='3.0.1', + version='3.1.0', author='John Good', author_email='jcg@caltech.edu', license='LICENSE', diff --git a/tests/_helpers.py b/tests/_helpers.py index 11873ce..60c32c1 100644 --- a/tests/_helpers.py +++ b/tests/_helpers.py @@ -20,3 +20,25 @@ def row_count_from_ipac(body: str) -> int: """ lines = [L for L in body.splitlines() if L.strip()] return max(0, len(lines) - 4) + + +def raw_get(server: str, path: str) -> bytes: + """GET *path* and return the response bytes exactly as sent (no parsing). + + HTTP libraries reject or repair a response with no status line, which is + the very thing the VOSI tests need to see. + """ + import socket + from urllib.parse import urlsplit + + u = urlsplit(server) + with socket.create_connection((u.hostname, u.port), timeout=30) as s: + s.sendall((f"GET {path} HTTP/1.1\r\nHost: {u.netloc}\r\n" + f"Connection: close\r\n\r\n").encode()) + chunks = [] + while True: + b = s.recv(65536) + if not b: + break + chunks.append(b) + return b"".join(chunks) diff --git a/tests/conftest.py b/tests/conftest.py index 6aa0cf6..e2e9b20 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -50,7 +50,9 @@ def fixture_root(tmp_path_factory) -> Path: build_test_db.build_all(str(root)) - (root / "workdir").mkdir() + # tap.py resolves a VOSI request's workspace as /TAP and + # fails with a 500 if it is missing; a deployed service always has it. + (root / "workdir" / "TAP").mkdir(parents=True) cgi_dir = root / "cgi-bin" / "TAP" cgi_dir.mkdir(parents=True) shutil.copy(FIXTURES / "nph-tap.py", cgi_dir / "nph-tap.py") @@ -66,7 +68,8 @@ def tap_conf(fixture_root: Path) -> Path: template = (FIXTURES / "TAP.conf.template").read_text() conf = template.format( TEST_WORKDIR=str(fixture_root / "workdir"), - TEST_HTTP_URL=f"http://127.0.0.1:{port}", + TEST_HTTP_HOST="127.0.0.1", + TEST_HTTP_PORT=str(port), TEST_DB_PATH=str(fixture_root / "test_data.db"), TEST_TAP_SCHEMA=str(fixture_root / "tap_schema.db"), ) @@ -136,6 +139,10 @@ def _run_cgi(self): "CONTENT_LENGTH": str(content_length), "HTTP_HOST": self.headers.get("Host", ""), }) + # A server may carry its own TAP.conf (legacy_tap_server does); + # otherwise the process-wide TAP_CONF set by tap_server applies. + if getattr(self.server, "tap_conf", None): + env["TAP_CONF"] = self.server.tap_conf import subprocess proc = subprocess.Popen( @@ -147,6 +154,12 @@ def _run_cgi(self): ) stdout, stderr = proc.communicate(input=body) + # A CGI that dies has only stderr to say so, and swallowing it + # makes a 500 from the script indistinguishable from a 500 the + # script meant to send. Surface it on the test runner's stderr. + if stderr: + sys.stderr.write(stderr.decode("utf-8", "replace")) + # Forward raw subprocess stdout to the socket. The script is # responsible for emitting a valid HTTP status line + headers. try: @@ -206,3 +219,37 @@ def tap_server(fixture_root: Path, tap_conf: Path): os.chdir(original_cwd) +@pytest.fixture(scope="session") +def legacy_tap_server(fixture_root: Path, tap_server: str): + """The same SQLite data behind NEA's 1.x TAP.conf layout: compat mode on. + + Depends on tap_server for PYTHONPATH and the working directory it sets. + """ + port = _free_port() + workdir = fixture_root / "workdir-legacy" + (workdir / "TAP").mkdir(parents=True, exist_ok=True) # as in fixture_root + conf = (FIXTURES / "TAP.conf.legacy.template").read_text().format( + TEST_WORKDIR=str(workdir), + TEST_HTTP_HOST="127.0.0.1", + TEST_HTTP_PORT=str(port), + TEST_DB_PATH=str(fixture_root / "test_data.db"), + TEST_TAP_SCHEMA=str(fixture_root / "tap_schema.db"), + ) + conf_path = fixture_root / "TAP.legacy.conf" + conf_path.write_text(conf) + httpd = http.server.HTTPServer(("127.0.0.1", port), _NphCGIHandler) + httpd.allow_reuse_address = True + httpd.nph_script_path = str(fixture_root / "cgi-bin" / "TAP" / "nph-tap.py") + httpd.tap_conf = str(conf_path) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + for _ in range(50): + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.2): + break + except OSError: + time.sleep(0.05) + try: + yield f"http://127.0.0.1:{port}" + finally: + httpd.shutdown() diff --git a/tests/fixtures/TAP.conf.legacy.template b/tests/fixtures/TAP.conf.legacy.template new file mode 100644 index 0000000..8d414c3 --- /dev/null +++ b/tests/fixtures/TAP.conf.legacy.template @@ -0,0 +1,23 @@ +# NEA's 1.x TAP.conf layout ([webserver] + [sqlite3] + ADQL_*), for the +# compat-mode tests. Same SQLite data as TAP.conf.template; placeholders +# are filled by the legacy_tap_server fixture in tests/conftest.py. +[webserver] +DBMS=sqlite3 +TAP_WORKDIR={TEST_WORKDIR} +TAP_WORKURL=/workspace +HTTP_URL=http://{TEST_HTTP_HOST} +HTTP_PORT={TEST_HTTP_PORT} +CGI_PGM=TAP +RACOL=ra +DECCOL=dec +ADQL_MODE=HTM +ADQL_LEVEL=20 +ADQL_XCOL=x +ADQL_YCOL=y +ADQL_ZCOL=z +ADQL_COLNAME=sptind +ADQL_ENCODING=BASE10 + +[sqlite3] +DB={TEST_DB_PATH} +TAP_SCHEMA={TEST_TAP_SCHEMA} diff --git a/tests/fixtures/TAP.conf.template b/tests/fixtures/TAP.conf.template index e0280d2..6c2c4a9 100644 --- a/tests/fixtures/TAP.conf.template +++ b/tests/fixtures/TAP.conf.template @@ -5,7 +5,8 @@ # # Placeholders: # {TEST_WORKDIR} — working dir for TAP request artifacts -# {TEST_HTTP_URL} — http://localhost:NNNN (fixture-allocated) +# {TEST_HTTP_HOST} — 127.0.0.1 +# {TEST_HTTP_PORT} — fixture-allocated port # {TEST_DB_PATH} — absolute path to test_data.db # {TEST_TAP_SCHEMA} — absolute path to tap_schema.db @@ -13,8 +14,12 @@ TAP_WORKDIR={TEST_WORKDIR} TAP_WORKURL=/workspace - HTTP_URL={TEST_HTTP_URL} - HTTP_PORT=0 + # configparam appends HTTP_PORT to HTTP_URL for any port other than + # 80/443, and that composed value is what the async status and result + # URLs are built from. Keep the two apart here or the job URLs come + # back with the port twice. + HTTP_URL=http://{TEST_HTTP_HOST} + HTTP_PORT={TEST_HTTP_PORT} CGI_PGM=TAP diff --git a/tests/test_async_flow.py b/tests/test_async_flow.py new file mode 100644 index 0000000..b8e2201 --- /dev/null +++ b/tests/test_async_flow.py @@ -0,0 +1,212 @@ +"""End-to-end coverage of the UWS async flow. + +An async TAP job takes three round trips, the same sequence pyvo and +pyNEID drive: + +1. POST to ``/async`` with the query. The server creates the job in + PENDING and answers 303 with ``Location: ``. +2. POST ``PHASE=RUN`` to ``/phase``. The server starts the + job and answers 303 to the same status URL. +3. GET ```` until the phase reaches COMPLETED, then read the + result named in the status document. + +None of this could be covered before the detach fix. ``tap.py`` ended +every async response by SIGKILLing ``os.getppid()`` — under this fixture +that is the process running the test suite, and in production it is the +web server child serving the request. Step 2 therefore either hung (the +CGI never closed the stdout pipe the server was reading) or took the +test runner down with it. +""" +from __future__ import annotations + +import time +import xml.etree.ElementTree as ET +from pathlib import Path + +import requests +from _helpers import tap_async_url + +UWS = {"uws": "http://www.ivoa.net/xml/UWS/v1.0", + "xlink": "http://www.w3.org/1999/xlink"} + +QUERY = "select * from data_l0" + + +def _submit(server: str) -> str: + """Round trip 1. Returns the job's status URL.""" + resp = requests.post( + tap_async_url(server), + data={"query": QUERY, "format": "ipac"}, + allow_redirects=False, + timeout=30, + ) + assert resp.status_code == 303, resp.text[:500] + statusurl = resp.headers["Location"] + assert statusurl, "303 carried no Location header" + return statusurl + + +def _status(statusurl: str) -> ET.Element: + resp = requests.get(statusurl, timeout=30) + assert resp.status_code == 200, resp.text[:500] + return ET.fromstring(resp.text) + + +def _phase(statusurl: str) -> str: + return _status(statusurl).find("uws:phase", UWS).text + + +def test_async_submit_creates_pending_job(tap_server): + statusurl = _submit(tap_server) + assert _phase(statusurl) == "PENDING" + + +def test_async_run_starts_the_job(tap_server): + """Round trip 2 answers promptly and does not take the server with it.""" + statusurl = _submit(tap_server) + + resp = requests.post( + f"{statusurl}/phase", + data={"PHASE": "RUN"}, + allow_redirects=False, + timeout=30, + ) + + assert resp.status_code == 303, resp.text[:500] + assert resp.headers["Location"] == statusurl + + +def test_async_run_response_is_self_delimiting(tap_server): + """The 303 must be framed by its headers, not by the connection dying. + + A reverse proxy in front of the CGI needs a Content-Length (or a + chunked body) to relay the response. The old code emitted a body + with neither and relied on the connection being destroyed to mark + the end of the message. + """ + statusurl = _submit(tap_server) + + resp = requests.post( + f"{statusurl}/phase", + data={"PHASE": "RUN"}, + allow_redirects=False, + timeout=30, + ) + + assert "Content-Length" in resp.headers + assert int(resp.headers["Content-Length"]) == len(resp.content) + + +def test_async_job_runs_to_completion(tap_server, fixture_root: Path): + statusurl = _submit(tap_server) + requests.post( + f"{statusurl}/phase", + data={"PHASE": "RUN"}, + allow_redirects=False, + timeout=30, + ) + + deadline = time.time() + 30 + phase = "" + while time.time() < deadline: + phase = _phase(statusurl) + if phase in ("COMPLETED", "ERROR", "ABORTED"): + break + time.sleep(0.2) + + assert phase == "COMPLETED", f"job ended in phase {phase}" + + job = _status(statusurl) + href = job.find("uws:results/uws:result", UWS).get( + "{http://www.w3.org/1999/xlink}href") + assert href + + # The fixture server does not serve TAP_WORKURL (Apache does that in + # production), so read the result out of the workspace on disk. + # tap.py puts each job under /TAP/. + workspace = job.find("uws:jobId", UWS).text + results = list((fixture_root / "workdir" / "TAP" / workspace).glob("*")) + assert results, f"no result artifacts in workspace {workspace}" + written = [p for p in results if p.name != "status.xml" and p.stat().st_size] + assert written, f"job completed but wrote no result file: {results}" + + +def _post_phase(statusurl: str, phase: str): + return requests.post( + f"{statusurl}/phase", + data={"PHASE": phase}, + allow_redirects=False, + timeout=30, + ) + + +def _results(fixture_root: Path, workspace: str) -> list: + workdir = fixture_root / "workdir" / "TAP" / workspace + return [p for p in workdir.glob("*") if p.name != "status.xml"] + + +def test_async_abort_is_terminal(tap_server, fixture_root: Path): + """ABORT must end the job, not run it. + + Every phase shares one response path at the end of the async block, + and that path used to fall through into the query. So a POST of + PHASE=ABORT wrote ABORTED, answered the client, and then ran the + job anyway, overwriting the phase the client had just been told. + """ + statusurl = _submit(tap_server) + assert _phase(statusurl) == "PENDING" + + resp = _post_phase(statusurl, "ABORT") + assert resp.status_code == 303, resp.text[:500] + + job = _status(statusurl) + assert job.find("uws:phase", UWS).text == "ABORTED" + + workspace = job.find("uws:jobId", UWS).text + time.sleep(1.0) + assert _phase(statusurl) == "ABORTED", "the aborted job kept running" + assert not _results(fixture_root, workspace), \ + "the aborted job wrote a result" + + +def test_async_unknown_phase_is_terminal(tap_server, fixture_root: Path): + """An unrecognized phase reports ERROR and runs nothing.""" + statusurl = _submit(tap_server) + + resp = _post_phase(statusurl, "SPIN") + assert resp.status_code == 303, resp.text[:500] + + job = _status(statusurl) + assert job.find("uws:phase", UWS).text == "ERROR" + + workspace = job.find("uws:jobId", UWS).text + time.sleep(1.0) + assert _phase(statusurl) == "ERROR", "the rejected job kept running" + assert not _results(fixture_root, workspace), \ + "the rejected job wrote a result" + + +def test_async_detach_does_not_signal_the_web_server(): + """Regression guard against the SIGKILL-the-parent detach. + + Asserted on the source rather than on behavior on purpose: the + behavioral assertion is "the process running this suite is still + alive", which a suite that has been SIGKILLed cannot make. + """ + import tokenize + + from TAP import tap + + # Comments and strings are stripped before the check so the code can + # still explain in prose why it no longer signals its parent. + with open(tap.__file__, "rb") as fp: + src = "".join( + tok.string + for tok in tokenize.tokenize(fp.readline) + if tok.type not in (tokenize.COMMENT, tokenize.STRING)) + + assert "getppid" not in src, ( + "tap.py signals its parent process. Under CGI the parent is the " + "web server child serving the request; killing it drops the " + "response and, behind a proxy, poisons the upstream connection." + ) diff --git a/tests/test_compat_config.py b/tests/test_compat_config.py new file mode 100644 index 0000000..ef749d4 --- /dev/null +++ b/tests/test_compat_config.py @@ -0,0 +1,135 @@ +"""Compat mode's config side: COMPAT names and the 1.x TAP.conf translator.""" +import pytest + +from TAP import compat +from TAP.configparam import configParam + +LEGACY_ORACLE = """[webserver] +DBMS=oracle +TAP_WORKDIR={work}/ +TAP_WORKURL=/workspace/ +HTTP_PORT=80 +COOKIENAME=tapcookie +CGI_PGM=/TAP +RACOL=ra +DECCOL=dec +# Spatial indexing settings +ADQL_MODE=HTM +ADQL_LEVEL=20 +ADQL_XCOL=x +ADQL_YCOL=y +ADQL_ZCOL=z +ADQL_COLNAME=htm20 +ADQL_ENCODING=BASE10 +HTTP_URL=https://exoplanetarchive.ipac.caltech.edu +[oracle] +ServerName=tapdb +UserID=exo_tap +Password=secret +""" + +CURRENT_ORACLE = """[WEB] +TAP_WORKDIR={work}/ +TAP_WORKURL=/workspace/ +HTTP_URL=https://exoplanetarchive.ipac.caltech.edu +HTTP_PORT=80 +CGI_PGM=/TAP +{compat} +[DBMS] +DBMS=oracle +ServerName=tapdb +UserID=exo_tap +Password=secret +COOKIENAME=tapcookie +RACOL=ra +DECCOL=dec +[SPTIND] +MODE=HTM +LEVEL=20 +XCOL=x +YCOL=y +ZCOL=z +COLNAME=htm20 +ENCODING=BASE10 +""" + + +def write(tmp_path, name, text, **fmt): + p = tmp_path / name + p.write_text(text.format(work=tmp_path, **fmt)) + return str(p) + + +def state(cp): + """Everything configParam derived, minus where it came from.""" + out = {k: v for k, v in vars(cp).items() if k not in ('configpath', 'compat', 'connectInfo')} + out['connectInfo'] = {k: v for k, v in cp.connectInfo.items() if k != 'compat'} + return out + + +def test_parse_names_accepts_str_and_list(): + assert compat.parse_names(None) == compat.NONE + assert compat.parse_names('nea-errors') == {'nea-errors'} + assert compat.parse_names(['nea-errors', ' nea-uws ']) == {'nea-errors', 'nea-uws'} + assert compat.parse_names(['', '']) == compat.NONE + + +def test_unknown_compat_name_is_an_error(tmp_path): + with pytest.raises(compat.CompatConfigError, match='nea-eror'): + compat.parse_names(['nea-eror']) + path = write(tmp_path, 'TAP.conf', CURRENT_ORACLE, compat='COMPAT = nea-eror') + with pytest.raises(compat.CompatConfigError): + configParam(path) + + +def test_legacy_oracle_reads_like_its_3x_equivalent(tmp_path): + legacy = configParam(write(tmp_path, 'legacy.conf', LEGACY_ORACLE)) + current = configParam(write(tmp_path, 'current.conf', CURRENT_ORACLE, compat='')) + assert state(legacy) == state(current) + assert legacy.compat == compat.ALL + assert legacy.connectInfo['compat'] == compat.ALL + assert current.compat == compat.NONE + assert current.connectInfo['compat'] == compat.NONE + + +def test_explicit_compat_list(tmp_path): + cp = configParam(write(tmp_path, 'TAP.conf', CURRENT_ORACLE, + compat='COMPAT = nea-errors, nea-uws')) + assert cp.compat == {'nea-errors', 'nea-uws'} + + +def test_adql_keys_after_db_section_are_found(tmp_path): + # configobj files keys that follow [oracle] under [oracle]. + text = LEGACY_ORACLE.replace('ADQL_MODE=HTM\n', '').replace( + 'Password=secret\n', 'Password=secret\nADQL_MODE=HTM\n') + conf = compat.translate_legacy(__import__('configobj').ConfigObj( + write(tmp_path, 'TAP.conf', text))) + assert conf['SPTIND']['MODE'] == 'HTM' + assert 'ADQL_MODE' not in conf['DBMS'] + + +def test_legacy_sqlite_gets_an_attach_name(tmp_path): + text = '[webserver]\nDBMS=sqlite3\nTAP_WORKDIR={work}/\n[sqlite3]\nDB=/x.db\nTAP_SCHEMA=/ts.db\n' + conf = compat.translate_legacy(__import__('configobj').ConfigObj( + write(tmp_path, 'TAP.conf', text))) + assert dict(conf['DBMS']) == {'DBMS': 'sqlite3', 'DB': '/x.db', 'TAP_SCHEMA': '/ts.db', + 'TAP_SCHEMA_FILE': 'TAP_SCHEMA'} + + +def test_legacy_without_its_db_section_is_an_error(tmp_path): + text = '[webserver]\nDBMS=oracle\nTAP_WORKDIR={work}/\n' + with pytest.raises(compat.CompatConfigError, match=r'no \[oracle\] section'): + compat.translate_legacy(__import__('configobj').ConfigObj( + write(tmp_path, 'TAP.conf', text))) + + +def test_compat_outside_web_is_an_error(tmp_path): + text = CURRENT_ORACLE.replace('Password=secret\n', 'Password=secret\nCOMPAT = nea-uws\n') + with pytest.raises(compat.CompatConfigError, match=r'\[DBMS\]'): + configParam(write(tmp_path, 'TAP.conf', text, compat='')) + + +def test_compat_in_legacy_webserver_is_an_error(tmp_path): + text = LEGACY_ORACLE.replace('DBMS=oracle\n', 'DBMS=oracle\nCOMPAT=nea-uws\n', 1) + with pytest.raises(compat.CompatConfigError, match=r'\[webserver\]'): + configParam(write(tmp_path, 'TAP.conf', text)) diff --git a/tests/test_compat_http.py b/tests/test_compat_http.py new file mode 100644 index 0000000..b49556b --- /dev/null +++ b/tests/test_compat_http.py @@ -0,0 +1,179 @@ +"""Compat mode end to end on the SQLite fixture: a 1.x TAP.conf gives NEA's +responses (legacy_tap_server); the 3.x config is unchanged (tap_server).""" +from __future__ import annotations + +import time +import xml.etree.ElementTree as ET + +import requests +from _helpers import raw_get, tap_async_url, tap_sync_url + +VOSI_HEAD = (b"HTTP/1.1 200 OK\r\nContent-type: application/xml\r\n" + b"Connection: close\r\n\r\n") +XML_DECL = b'\n' + + +def test_vosi_documents_carry_neas_headers(legacy_tap_server): + for ep in ("availability", "capabilities"): + raw = raw_get(legacy_tap_server, "/cgi-bin/TAP/nph-tap.py/" + ep) + assert raw.startswith(VOSI_HEAD + XML_DECL), raw[:200] + assert raw.endswith(b"\n\r\n"), raw[-80:] + + +def test_vosi_documents_unchanged_in_default_mode(tap_server): + for ep in ("availability", "capabilities"): + raw = raw_get(tap_server, "/cgi-bin/TAP/nph-tap.py/" + ep) + assert raw.startswith(XML_DECL), raw[:200] + + +DENIED = "Table 'not_a_table' is not available for querying." + + +def _sync(server, query, fmt="csv"): + return requests.post(tap_sync_url(server), data={"query": query, "format": fmt}, + timeout=60) + + +def test_unknown_table_is_neas_400_votable(legacy_tap_server): + r = _sync(legacy_tap_server, "select * from not_a_table") + assert r.status_code == 400 + assert r.headers["Content-Type"] == "application/xml" + assert '\n' + DENIED in r.text + + +def test_bad_adql_is_a_votable_error(legacy_tap_server): + r = _sync(legacy_tap_server, "selec x frm y") + assert r.status_code == 400 + assert r.headers["Content-Type"] == "application/xml" + assert r.text.startswith('\n" not in r.text + assert '' in r.text + + +def test_votable_unchanged_in_default_mode(tap_server): + r = _sync(tap_server, NONCHAR, fmt="votable") + assert r.status_code == 200, r.text[:300] + assert r.headers["Content-Type"] == "text/xml" + assert "Julian date of observation." in r.text + + +UWS = "{http://www.ivoa.net/xml/UWS/v1.0}" + + +def _completed_job(server): + """Submit, run and wait; returns (the 303 answering RUN, status URL). + + The 303 to the initial submit (job left PENDING) is written by a different + code path and carries no Content-Type in either mode; the one answering + PHASE=RUN is what ``async_submit_headers`` controls. + """ + sub = requests.post(tap_async_url(server), + data={"query": "select obsjd from data_l0", "format": "csv"}, + allow_redirects=False, timeout=60) + assert sub.status_code == 303, sub.text[:300] + url = sub.headers["Location"] + run = requests.post(url + "/phase", data={"PHASE": "RUN"}, + allow_redirects=False, timeout=60) + assert run.status_code == 303, run.text[:300] + deadline = time.time() + 30 + while time.time() < deadline: + if requests.get(url + "/phase", timeout=30).text.strip() in ("COMPLETED", "ERROR"): + break + time.sleep(0.2) + return run, url + + +def test_uws_is_neas_form(legacy_tap_server): + run, url = _completed_job(legacy_tap_server) + assert "Content-Type" not in run.headers + assert "Content-Length" not in run.headers + assert "Connection" not in run.headers + r = requests.get(url, timeout=30) + assert r.headers["Content-Type"] == "application/xml" + job = ET.fromstring(r.text) + assert job.find(UWS + "phase").text == "COMPLETED" + assert job.find(UWS + "executionDuration") is not None + assert job.find(UWS + "executionduration") is None + assert not job.find(UWS + "destruction").text.endswith("Z") + dur = requests.get(url + "/executionduration", timeout=30) + assert dur.status_code == 200 and dur.text.strip().isdigit(), dur.text[:200] + camel = requests.get(url + "/executionDuration", timeout=30) + assert camel.status_code == 200 and camel.text.strip().isdigit(), camel.text[:200] + assert camel.text == dur.text + + +def test_uws_unchanged_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) + assert run.headers["Connection"] == "close" + 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 + "destruction").text.endswith("Z") + + +def test_camelcase_duration_key_unchanged_in_default_mode(tap_server): + run, url = _completed_job(tap_server) + r = requests.get(url + "/executionDuration", timeout=30) + assert r.status_code == 200 + assert r.headers["Content-Type"] == "text/plain" + assert r.text == "key executionDuration is not a valid key\n" + + +def _respell_duration(fixture_root, url, old, new): + """Rewrite one job's status.xml as if the other mode had written it.""" + jobid = url.rstrip("/").rsplit("/", 1)[1] + paths = list(fixture_root.glob("workdir*/**/%s/status.xml" % jobid)) + assert len(paths) == 1, paths + text = paths[0].read_text() + assert "uws:" + old in text + paths[0].write_text(text.replace("uws:" + old, "uws:" + new)) + + +def _read_back(url): + whole = requests.get(url, timeout=30) + assert whole.status_code == 200 and "COMPLETED<" in whole.text, whole.text[:300] + dur = requests.get(url + "/executionduration", timeout=30) + assert dur.status_code == 200 and dur.text.strip().isdigit(), dur.text[:300] + phase = requests.get(url + "/phase", timeout=30) + assert phase.text.strip() == "COMPLETED", phase.text[:300] + + +def test_status_written_by_default_mode_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): + _, url = _completed_job(tap_server) + _respell_duration(fixture_root, url, "executionduration", "executionDuration") + _read_back(url) + + +def test_tables_is_neas_layout(legacy_tap_server): + r = requests.get(legacy_tap_server + "/cgi-bin/TAP/nph-tap.py/tables", timeout=60) + assert r.status_code == 200, r.text[:300] + + +def test_tables_unchanged_in_default_mode(tap_server): + r = requests.get(tap_server + "/cgi-bin/TAP/nph-tap.py/tables", timeout=60) + assert r.status_code == 200, r.text[:300] + assert "principal" not in r.text diff --git a/tests/test_compat_units.py b/tests/test_compat_units.py new file mode 100644 index 0000000..7059818 --- /dev/null +++ b/tests/test_compat_units.py @@ -0,0 +1,97 @@ +"""compat.py's pure functions, active and inactive.""" +import datetime +import io + +from TAP import compat + + +def test_error_document_escapes_markup_not_quotes(): + doc = compat.error_document(compat.ALL, "a < b & 'c'", '400') + assert doc.startswith('HTTP/1.1 400 ERROR\r\nContent-type: application/xml\r\n\r\n' + '\n') + assert "\na < b & 'c'\n\n" in doc + assert doc.endswith('\n\n\n') + + +def test_error_document_inactive_is_none(): + assert compat.error_document(compat.NONE, 'x', '400') is None + + +def test_table_denied_status(): + assert compat.table_denied_status(compat.ALL) == '400' + assert compat.table_denied_status(compat.NONE) == '403' + + +def test_votable_content_type(): + assert compat.votable_content_type(compat.ALL) == 'application/xml' + assert compat.votable_content_type(compat.NONE) == 'text/xml' + + +def test_field_description_blanks_only_non_char_columns(): + assert compat.field_description(compat.ALL, 'Julian date', 'double') == '' + assert compat.field_description(compat.ALL, 'Program ID', 'char') == 'Program ID' + assert compat.field_description(compat.NONE, 'Julian date', 'double') == 'Julian date' + + +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' + assert compat.uws_url_key(compat.ALL, 'executionDuration') == 'executionduration' + assert compat.uws_url_key(compat.ALL, 'executionduration') == 'executionduration' + assert compat.uws_url_key(compat.ALL, 'phase') == 'phase' + assert compat.uws_url_key(compat.NONE, 'executionDuration') == 'executionDuration' + assert compat.uws_url_key(compat.NONE, 'executionduration') == 'executionduration' + assert compat.destruction_suffix(compat.ALL) == '' + assert compat.destruction_suffix(compat.NONE) == 'Z' + assert compat.uws_content_type(compat.ALL) == 'application/xml' + assert compat.uws_content_type(compat.NONE) == 'text/xml' + body = 'Redirect Location: https://x/\u00e9\n' + assert len(body.encode('utf-8')) != len(body) + assert compat.async_submit_headers(compat.ALL, body) == '' + assert compat.async_submit_headers(compat.NONE, body) == ( + 'Content-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n' + % len(body.encode('utf-8'))) + + +def test_vosi_tables_class(): + from TAP import compat_vositables, vositables + assert compat.vosi_tables_class(compat.ALL) is compat_vositables.vosiTables + assert compat.vosi_tables_class(compat.NONE) is vositables.vosiTables + + +def test_warning_once_per_day(tmp_path): + out = io.StringIO() + day1, day2 = datetime.date(2026, 10, 6), datetime.date(2026, 10, 7) + assert compat.warn_once_per_day(compat.ALL, str(tmp_path), day1, out) is True + assert compat.warn_once_per_day(compat.ALL, str(tmp_path), day1, out) is False + assert compat.warn_once_per_day(compat.ALL, str(tmp_path), day2, out) is True + lines = out.getvalue().splitlines() + assert len(lines) == 2 + assert lines[0].startswith('nexsciTAP: compatibility mode active (nea-errors, ') + assert lines[0].endswith('see MIGRATING.md') + assert (tmp_path / 'TAP' / '.compat-warned-2026-10-06').exists() + + +def test_no_warning_without_compat(tmp_path): + out = io.StringIO() + assert compat.warn_once_per_day(compat.NONE, str(tmp_path), None, out) is False + assert out.getvalue() == '' and not (tmp_path / 'TAP').exists() + + +def test_warning_never_fails_a_request(tmp_path): + blocker = tmp_path / 'file' # a file where the work dir should be + blocker.write_text('x') + out = io.StringIO() + assert compat.warn_once_per_day(compat.ALL, str(blocker), None, out) is False + assert compat.warn_once_per_day(compat.ALL, None, None, out) is False + + +def test_duration_value_reads_either_spelling(): + lower = {'uws:executionduration': '30'} + camel = {'uws:executionDuration': '30'} + assert compat.duration_value(lower) == '30' + assert compat.duration_value(camel) == '30' + assert compat.duration_value({}) is None diff --git a/tests/test_pyneid_compat.py b/tests/test_pyneid_compat.py index e927e0a..883a3b7 100644 --- a/tests/test_pyneid_compat.py +++ b/tests/test_pyneid_compat.py @@ -1,54 +1,40 @@ """pyNEID compatibility regression tests. -Currently all skipped — the test fixture's HTTP handler doesn't yet -support TAP's full async flow end-to-end. +Still skipped, but for a different reason than before. -Why these matter: pyNEID's `query_adql` always uses the async TAP -endpoint, which works in three round-trips: +pyNEID's ``query_adql`` always uses the async TAP endpoint, which works +in three round trips: -1. POST to /TAP/async with the ADQL query → server returns 303 See Other +1. POST to /TAP/async with the ADQL query -> server returns 303 See Other with a Location: header. -2. POST to with PHASE=RUN → server starts the job, returns - another 303 to the same status URL. +2. POST to /phase with PHASE=RUN -> server starts the job and + returns another 303 to the same status URL. 3. GET the status URL repeatedly until phase=COMPLETED, then GET the - results URL embedded in the status XML. - -Steps 2–3 depend on the server persisting per-request workspace state -(status.xml, results) under TAP_WORKDIR keyed by a workspace ID -embedded in the URL path. The current `_NphCGIHandler` in -tests/conftest.py routes by URL prefix but doesn't model the workspace -lifecycle — each request is a fresh CGI subprocess that doesn't know -about prior request state. Production runs under Apache, which manages -the same workdir mount across requests so the state naturally -persists, but reproducing that fidelity in CI needs: - -- Sticky working directory across requests (already mostly there since - we set `os.chdir(fixture_root)` once at fixture setup) -- Correct PATH_INFO routing for the second and subsequent calls in the - async flow (the Location URL pyNEID gets back may not match the URL - pattern the handler routes today) -- Possibly more debug — when this was tried locally, pyNEID hung at - step 2 with no observable error in the TAP debug log - -Equivalent direct-HTTP coverage of the same code paths exists in -tests/test_security_e2e.py and tests/test_http_endpoint.py — those -exercise the sync endpoint, which uses a single request and doesn't -need the workspace dance. The pyNEID layer adds value as a regression -check on top of those (it's the exact wrapper production traffic hits), -but is not blocking. - -To wire this up later: trace the async POST/poll loop, fix the handler -routing, then drop the skip decorator. + result URL embedded in the status XML. + +This module used to say the fixture handler could not carry that +sequence, and recorded that "pyNEID hung at step 2 with no observable +error in the TAP debug log". The hang was not the fixture. Step 2 ended +by SIGKILLing os.getppid(), which under the fixture is the test runner +and in production is the web server child serving the request, so the +client was left waiting on a pipe nobody would close. That is fixed, and +the sequence above now has direct coverage in tests/test_async_flow.py. + +What is still missing here is pyNEID itself: it is not a test dependency +and is not installed in CI, so there is nothing to import. Adding it +means weighing a heavier test dependency (and its own network defaults) +against coverage that tests/test_async_flow.py already provides at the +HTTP level. Until that call is made, this module stays skipped. + +To wire it up later: add pyNEID to requirements-test.txt, write the +round-trip test against the fixture server, and drop the skip below. """ from __future__ import annotations import pytest -# Once the async fixture support lands, drop this module-level skip and -# re-add the actual test bodies (a snapshot of the intended tests is in -# the git history of this file or in the design plan). pytest.skip( - "Async TAP flow not yet supported by the test fixture handler. " - "See module docstring for what's needed.", + "pyNEID is not a test dependency; the async flow it drives is covered " + "directly in tests/test_async_flow.py. See module docstring.", allow_module_level=True, ) diff --git a/tests/test_tap_schema_name.py b/tests/test_tap_schema_name.py new file mode 100644 index 0000000..87e34c0 --- /dev/null +++ b/tests/test_tap_schema_name.py @@ -0,0 +1,68 @@ +"""TAP_SCHEMA's name in SQL: the schema for Oracle, the ATTACH name for SQLite. + +configparam always passes tap_schema_file='TAP_SCHEMA.db' (SQLite's attach +name). Using it for Oracle produced 'FROM TAP_SCHEMA.db.tables' and +ORA-03048 on every query (found by comparing against a production deployment, 2026-10-05). +""" +import contextlib + +from TAP.datadictionary import dataDictionary +from TAP.tablevalidator import TableValidator + + +class _Cursor: + def __init__(self, log): + self.log = log + self.description = [('column_name',), ('datatype',), ('description',), ('unit',)] + self.arraysize = 100 + + def execute(self, sql, *params): + self.log.append(sql) + + def fetchall(self): + return [('ps',)] + + def fetchmany(self, *a): + return [] + + def close(self): + pass + + +class _Conn: + def __init__(self): + self.log = [] + + def cursor(self): + return _Cursor(self.log) + + +def info(dbms, file_name): + return {'dbms': dbms, 'tap_schema': 'TAP_SCHEMA', 'tap_schema_file': file_name, + 'tables_table': 'tables', 'columns_table': 'columns'} + + +def test_validator_uses_schema_name_for_oracle(): + conn = _Conn() + TableValidator(conn, info('oracle', 'TAP_SCHEMA.db')) + assert conn.log == ['SELECT table_name FROM TAP_SCHEMA.tables'] + + +def test_validator_uses_attach_name_for_sqlite(): + conn = _Conn() + TableValidator(conn, info('sqlite3', 'TS')) + assert conn.log == ['SELECT table_name FROM TS.tables'] + + +def test_data_dictionary_uses_schema_name_for_oracle(): + conn = _Conn() + with contextlib.suppress(Exception): # the fake has no rows; only the SQL matters + dataDictionary(conn, 'ps', info('oracle', 'TAP_SCHEMA.db')) + assert conn.log and conn.log[0].startswith('select * from TAP_SCHEMA.columns ') + + +def test_data_dictionary_uses_attach_name_for_sqlite(): + conn = _Conn() + with contextlib.suppress(Exception): + dataDictionary(conn, 'ps', info('sqlite3', 'TS')) + assert conn.log and conn.log[0].startswith('select * from TS.columns ')