| from __future__ import print_function |
| from lnt.util import NTEmailReport |
| from contextlib import closing |
| from lnt.util import logger |
| import collections |
| import datetime |
| import lnt.formats |
| import lnt.server.reporting.analysis |
| import lnt.testing |
| import os |
| |
| import tempfile |
| import time |
| |
| from lnt.server.db import fieldchange |
| |
| |
| def import_and_report(config, db_name, db, session, file, format, ts_name, |
| show_sample_count=False, disable_email=False, |
| disable_report=False, select_machine=None, |
| merge_run=None): |
| """ |
| import_and_report(config, db_name, db, session, file, format, ts_name, |
| [show_sample_count], [disable_email], |
| [disable_report], [select_machine], [merge_run]) |
| -> ... object ... |
| |
| Import a test data file into an LNT server and generate a test report. On |
| success, run is the newly imported run. |
| |
| The result object is a dictionary containing information on the imported |
| run and its comparison to the previous run. |
| """ |
| result = { |
| 'success': False, |
| 'error': None, |
| 'import_file': file, |
| } |
| if select_machine is None: |
| select_machine = 'match' |
| if merge_run is None: |
| merge_run = 'reject' |
| |
| if select_machine not in ('match', 'update', 'split'): |
| result['error'] = "select_machine must be 'match', 'update' or 'split'" |
| return result |
| |
| ts = db.testsuite.get(ts_name, None) |
| if ts is None: |
| result['error'] = "Unknown test suite '%s'!" % ts_name |
| return result |
| numMachines = ts.getNumMachines(session) |
| numRuns = ts.getNumRuns(session) |
| numTests = ts.getNumTests(session) |
| |
| # If the database gets fragmented, count(*) in SQLite can get really |
| # slow!?! |
| if show_sample_count: |
| numSamples = ts.getNumSamples(session) |
| |
| startTime = time.time() |
| try: |
| data = lnt.formats.read_any(file, format) |
| except Exception: |
| import traceback |
| result['error'] = "could not parse input format" |
| result['message'] = traceback.format_exc() |
| return result |
| |
| result['load_time'] = time.time() - startTime |
| |
| # Auto-upgrade the data, if necessary. |
| try: |
| data = lnt.testing.upgrade_and_normalize_report(data, ts_name) |
| except ValueError as e: |
| import traceback |
| result['error'] = "Invalid input format: %s" % e |
| result['message'] = traceback.format_exc() |
| return result |
| |
| # Find the database config, if we have a configuration object. |
| if config: |
| db_config = config.databases[db_name] |
| else: |
| db_config = None |
| |
| # Find the email address for this machine's results. |
| toAddress = email_config = None |
| if db_config and not disable_email: |
| email_config = db_config.email_config |
| if email_config.enabled: |
| # Find the machine name. |
| machineName = str(data.get('Machine', {}).get('Name')) |
| toAddress = email_config.get_to_address(machineName) |
| if toAddress is None: |
| result['error'] = ("unable to match machine name " |
| "for test results email address!") |
| return result |
| |
| importStartTime = time.time() |
| try: |
| data_schema = data.get('schema') |
| if data_schema is not None and data_schema != ts_name: |
| result['error'] = ("Importing '%s' data into test suite '%s'" % |
| (data_schema, ts_name)) |
| return result |
| |
| run = ts.importDataFromDict(session, data, config=db_config, |
| select_machine=select_machine, |
| merge_run=merge_run) |
| except KeyboardInterrupt: |
| raise |
| except Exception as e: |
| import traceback |
| result['error'] = "import failure: %s" % e |
| result['message'] = traceback.format_exc() |
| if isinstance(e, lnt.server.db.testsuitedb.MachineInfoChanged): |
| result['message'] += \ |
| '\n\nNote: Use --select-machine=update to update ' \ |
| 'the existing machine information.\n' |
| return result |
| |
| # If the import succeeded, save the import path. |
| run.imported_from = file |
| |
| result['import_time'] = time.time() - importStartTime |
| |
| result['report_to_address'] = toAddress |
| if config: |
| report_url = "%s/db_%s/" % (config.zorgURL, db_name) |
| else: |
| report_url = "localhost" |
| |
| if not disable_report: |
| # This has the side effect of building the run report for |
| # this result. |
| NTEmailReport.emailReport(result, session, run, report_url, |
| email_config, toAddress, True) |
| |
| result['added_machines'] = ts.getNumMachines(session) - numMachines |
| result['added_runs'] = ts.getNumRuns(session) - numRuns |
| result['added_tests'] = ts.getNumTests(session) - numTests |
| if show_sample_count: |
| result['added_samples'] = ts.getNumSamples(session) - numSamples |
| |
| result['committed'] = True |
| result['run_id'] = run.id |
| session.commit() |
| |
| fieldchange.post_submit_tasks(session, ts, run.id) |
| |
| # Add a handy relative link to the submitted run. |
| result['result_url'] = "db_{}/v4/{}/{}".format(db_name, ts_name, run.id) |
| result['report_time'] = time.time() - importStartTime |
| result['total_time'] = time.time() - startTime |
| logger.info("Successfully created {}".format(result['result_url'])) |
| # If this database has a shadow import configured, import the run into that |
| # database as well. |
| if config and config.databases[db_name].shadow_import: |
| # Load the shadow database to import into. |
| db_config = config.databases[db_name] |
| shadow_name = db_config.shadow_import |
| with closing(config.get_database(shadow_name)) as shadow_db: |
| if shadow_db is None: |
| raise ValueError("invalid configuration, shadow import " |
| "database %r does not exist" % shadow_name) |
| |
| # Perform the shadow import. |
| shadow_session = shadow_db.make_session() |
| shadow_result = import_and_report(config, shadow_name, |
| shadow_db, shadow_session, file, |
| format, ts_name, |
| show_sample_count, disable_email, |
| disable_report, |
| select_machine=select_machine, |
| merge_run=merge_run) |
| |
| # Append the shadow result to the result. |
| result['shadow_result'] = shadow_result |
| |
| result['success'] = True |
| return result |
| |
| |
| def no_submit(): |
| """Do not submit but create dummy submission report.""" |
| return { |
| 'success': True, |
| 'result_url': None, |
| 'test_results': None, |
| } |
| |
| |
| def print_report_result(result, out, err, verbose=True): |
| """ |
| print_report_result(result, out, [err], [verbose]) -> None |
| |
| Print a human readable form of an import result object to the given output |
| stream. Test results are printed in 'lit' format. |
| """ |
| |
| # Print the generic import information. |
| if 'import_file' in result: |
| print("Importing %r" % os.path.basename(result['import_file']), file=out) |
| if result['success']: |
| print("Import succeeded.", file=out) |
| print(file=out) |
| else: |
| out.flush() |
| print("Import Failed:", file=err) |
| print("%s\n" % result['error'], file=err) |
| message = result.get('message', None) |
| if message: |
| print("%s\n" % message, file=err) |
| print("--------------", file=err) |
| err.flush() |
| return |
| |
| # Print the test results. |
| test_results = result.get('test_results') |
| if not test_results: |
| return |
| |
| # List the parameter sets, if interesting. |
| show_pset = len(test_results) > 1 |
| if show_pset: |
| print("Parameter Sets", file=out) |
| print("--------------", file=out) |
| for i, info in enumerate(test_results): |
| print("P%d: %s" % (i, info['pset']), file=out) |
| print(file=out) |
| |
| total_num_tests = sum([len(item['results']) |
| for item in test_results]) |
| print("--- Tested: %d tests --" % total_num_tests, file=out) |
| test_index = 0 |
| result_kinds = collections.Counter() |
| for i, item in enumerate(test_results): |
| pset_results = item['results'] |
| |
| for name, test_status, perf_status in pset_results: |
| test_index += 1 |
| # FIXME: Show extended information for performance changes, |
| # previous samples, standard deviation, all that. |
| # |
| # FIXME: Think longer about mapping to test codes. |
| result_info = None |
| |
| if test_status == lnt.server.reporting.analysis.REGRESSED: |
| result_string = 'FAIL' |
| elif test_status == lnt.server.reporting.analysis.UNCHANGED_FAIL: |
| result_string = 'FAIL' |
| elif test_status == lnt.server.reporting.analysis.IMPROVED: |
| result_string = 'IMPROVED' |
| result_info = "Test started passing." |
| elif perf_status is None: |
| # Missing perf status means test was just added or removed. |
| result_string = 'PASS' |
| elif perf_status == lnt.server.reporting.analysis.REGRESSED: |
| result_string = 'REGRESSED' |
| result_info = 'Performance regressed.' |
| elif perf_status == lnt.server.reporting.analysis.IMPROVED: |
| result_string = 'IMPROVED' |
| result_info = 'Performance improved.' |
| else: |
| result_string = 'PASS' |
| result_kinds[result_string] += 1 |
| # Ignore passes unless in verbose mode. |
| if not verbose and result_string == 'PASS': |
| continue |
| |
| if show_pset: |
| name = 'P%d :: %s' % (i, name) |
| print("%s: %s (%d of %d)" % (result_string, name, |
| test_index, total_num_tests), file=out) |
| |
| if result_info: |
| print("%s TEST '%s' %s" % ('*'*20, name, '*'*20), file=out) |
| print(result_info, file=out) |
| print("*" * 20, file=out) |
| |
| if 'original_run' in result: |
| print(("This submission is a duplicate of run %d, " |
| "already in the database.") % result['original_run'], file=out) |
| print(file=out) |
| |
| if result['report_to_address']: |
| print("Report emailed to: %r" % result['report_to_address'], file=out) |
| print(file=out) |
| |
| # Print the processing times. |
| print("Processing Times", file=out) |
| print("----------------", file=out) |
| print("Load : %.2fs" % result['load_time'], file=out) |
| print("Import : %.2fs" % result['import_time'], file=out) |
| print("Report : %.2fs" % result['report_time'], file=out) |
| print("Total : %.2fs" % result['total_time'], file=out) |
| print(file=out) |
| |
| # Print the added database items. |
| total_added = (result['added_machines'] + result['added_runs'] + |
| result['added_tests'] + result.get('added_samples', 0)) |
| if total_added: |
| print("Imported Data", file=out) |
| print("-------------", file=out) |
| if result['added_machines']: |
| print("Added Machines: %d" % result['added_machines'], file=out) |
| if result['added_runs']: |
| print("Added Runs : %d" % result['added_runs'], file=out) |
| if result['added_tests']: |
| print("Added Tests : %d" % result['added_tests'], file=out) |
| if result.get('added_samples', 0): |
| print("Added Samples : %d" % result['added_samples'], file=out) |
| print(file=out) |
| print("Results", file=out) |
| print("----------------", file=out) |
| for kind, count in result_kinds.items(): |
| print(kind, ":", count, file=out) |
| |
| |
| def import_from_string(config, db_name, db, session, ts_name, data, |
| select_machine=None, merge_run=None): |
| # Stash a copy of the raw submission. |
| # |
| # To keep the temporary directory organized, we keep files in |
| # subdirectories organized by (database, year-month). |
| utcnow = datetime.datetime.utcnow() |
| tmpdir = os.path.join(config.tempDir, db_name, |
| "%04d-%02d" % (utcnow.year, utcnow.month)) |
| try: |
| os.makedirs(tmpdir) |
| except OSError: |
| pass |
| |
| # Save the file under a name prefixed with the date, to make it easier |
| # to use these files in cases we might need them for debugging or data |
| # recovery. |
| prefix = utcnow.strftime("data-%Y-%m-%d_%H-%M-%S") |
| fd, path = tempfile.mkstemp(prefix=prefix, suffix='.json', |
| dir=str(tmpdir), text=True) |
| with os.fdopen(fd, "w") as fp: |
| fp.write(data) |
| |
| # Import the data. |
| # |
| # FIXME: Gracefully handle formats failures and DOS attempts. We |
| # should at least reject overly large inputs. |
| |
| result = lnt.util.ImportData.import_and_report( |
| config, db_name, db, session, path, '<auto>', ts_name, |
| select_machine=select_machine, merge_run=merge_run) |
| return result |
