How to use test_end method in localstack

Best Python code snippet using localstack_python

test_update.py

Source:test_update.py Github

copy

Full Screen

1import json2import mock3import os4import pytest5import sys6from io import BytesIO7from .. import metadata, manifestupdate8from ..update.update import WPTUpdate9from ..update.base import StepRunner, Step10from mozlog import structuredlog, handlers, formatters11here = os.path.dirname(__file__)12sys.path.insert(0, os.path.join(here, os.pardir, os.pardir, os.pardir))13from manifest import manifest, item as manifest_item, utils14def rel_path_to_test_url(rel_path):15 assert not os.path.isabs(rel_path)16 return rel_path.replace(os.sep, "/")17def SourceFileWithTest(path, hash, cls, *args):18 path_parts = tuple(path.split("/"))19 path = utils.to_os_path(path)20 s = mock.Mock(rel_path=path, rel_path_parts=path_parts, hash=hash)21 test = cls("/foobar", path, "/", rel_path_to_test_url(path), *args)22 s.manifest_items = mock.Mock(return_value=(cls.item_type, [test]))23 return s24def tree_and_sourcefile_mocks(source_files):25 paths_dict = {}26 tree = []27 for source_file, file_hash, updated in source_files:28 paths_dict[source_file.rel_path] = source_file29 tree.append([source_file.rel_path, file_hash, updated])30 def MockSourceFile(tests_root, path, url_base, file_hash):31 return paths_dict[path]32 return tree, MockSourceFile33item_classes = {"testharness": manifest_item.TestharnessTest,34 "reftest": manifest_item.RefTest,35 "manual": manifest_item.ManualTest,36 "wdspec": manifest_item.WebDriverSpecTest,37 "conformancechecker": manifest_item.ConformanceCheckerTest,38 "visual": manifest_item.VisualTest,39 "support": manifest_item.SupportFile}40default_run_info = {"debug": False, "os": "linux", "version": "18.04", "processor": "x86_64", "bits": 64}41test_id = "/path/to/test.htm"42dir_id = "path/to/__dir__"43def reset_globals():44 metadata.prop_intern.clear()45 metadata.run_info_intern.clear()46 metadata.status_intern.clear()47def get_run_info(overrides):48 run_info = default_run_info.copy()49 run_info.update(overrides)50 return run_info51def update(tests, *logs, **kwargs):52 full_update = kwargs.pop("full_update", False)53 disable_intermittent = kwargs.pop("disable_intermittent", False)54 update_intermittent = kwargs.pop("update_intermittent", False)55 remove_intermittent = kwargs.pop("remove_intermittent", False)56 assert not kwargs57 id_test_map, updater = create_updater(tests)58 for log in logs:59 log = create_log(log)60 updater.update_from_log(log)61 update_properties = (["debug", "os", "version", "processor"],62 {"os": ["version"], "processor": ["bits"]})63 expected_data = {}64 metadata.load_expected = lambda _, __, test_path, *args: expected_data.get(test_path)65 for test_path, test_ids, test_type, manifest_str in tests:66 test_path = utils.to_os_path(test_path)67 expected_data[test_path] = manifestupdate.compile(BytesIO(manifest_str),68 test_path,69 "/",70 update_properties,71 update_intermittent,72 remove_intermittent)73 return list(metadata.update_results(id_test_map,74 update_properties,75 full_update,76 disable_intermittent,77 update_intermittent,78 remove_intermittent))79def create_updater(tests, url_base="/", **kwargs):80 id_test_map = {}81 m = create_test_manifest(tests, url_base)82 reset_globals()83 id_test_map = metadata.create_test_tree(None, m)84 return id_test_map, metadata.ExpectedUpdater(id_test_map, **kwargs)85def create_log(entries):86 data = BytesIO()87 if isinstance(entries, list):88 logger = structuredlog.StructuredLogger("expected_test")89 handler = handlers.StreamHandler(data, formatters.JSONFormatter())90 logger.add_handler(handler)91 for item in entries:92 action, kwargs = item93 getattr(logger, action)(**kwargs)94 logger.remove_handler(handler)95 else:96 data.write(json.dumps(entries).encode())97 data.seek(0)98 return data99def suite_log(entries, run_info=None):100 _run_info = default_run_info.copy()101 if run_info:102 _run_info.update(run_info)103 return ([("suite_start", {"tests": [], "run_info": _run_info})] +104 entries +105 [("suite_end", {})])106def create_test_manifest(tests, url_base="/"):107 source_files = []108 for i, (test, _, test_type, _) in enumerate(tests):109 if test_type:110 source_files.append(SourceFileWithTest(test, str(i) * 40, item_classes[test_type]))111 m = manifest.Manifest("")112 tree, sourcefile_mock = tree_and_sourcefile_mocks((item, None, True) for item in source_files)113 with mock.patch("manifest.manifest.SourceFile", side_effect=sourcefile_mock):114 m.update(tree)115 return m116def test_update_0():117 tests = [("path/to/test.htm", [test_id], "testharness",118 b"""[test.htm]119 [test1]120 expected: FAIL""")]121 log = suite_log([("test_start", {"test": "/path/to/test.htm"}),122 ("test_status", {"test": "/path/to/test.htm",123 "subtest": "test1",124 "status": "PASS",125 "expected": "FAIL"}),126 ("test_end", {"test": "/path/to/test.htm",127 "status": "OK"})])128 updated = update(tests, log)129 assert len(updated) == 1130 assert updated[0][1].is_empty131def test_update_1():132 tests = [("path/to/test.htm", [test_id], "testharness",133 b"""[test.htm]134 [test1]135 expected: ERROR""")]136 log = suite_log([("test_start", {"test": test_id}),137 ("test_status", {"test": test_id,138 "subtest": "test1",139 "status": "FAIL",140 "expected": "ERROR"}),141 ("test_end", {"test": test_id,142 "status": "OK"})])143 updated = update(tests, log)144 new_manifest = updated[0][1]145 assert not new_manifest.is_empty146 assert new_manifest.get_test(test_id).children[0].get("expected", default_run_info) == "FAIL"147def test_update_known_intermittent_1():148 tests = [("path/to/test.htm", [test_id], "testharness",149 b"""[test.htm]150 [test1]151 expected: PASS""")]152 log_0 = suite_log([("test_start", {"test": test_id}),153 ("test_status", {"test": test_id,154 "subtest": "test1",155 "status": "FAIL",156 "expected": "PASS"}),157 ("test_end", {"test": test_id,158 "status": "OK"})])159 log_1 = suite_log([("test_start", {"test": test_id}),160 ("test_status", {"test": test_id,161 "subtest": "test1",162 "status": "PASS",163 "expected": "PASS"}),164 ("test_end", {"test": test_id,165 "status": "OK"})])166 log_2 = suite_log([("test_start", {"test": test_id}),167 ("test_status", {"test": test_id,168 "subtest": "test1",169 "status": "PASS",170 "expected": "PASS"}),171 ("test_end", {"test": test_id,172 "status": "OK"})])173 updated = update(tests, log_0, log_1, log_2, update_intermittent=True)174 new_manifest = updated[0][1]175 assert not new_manifest.is_empty176 assert new_manifest.get_test(test_id).children[0].get(177 "expected", default_run_info) == ["PASS", "FAIL"]178def test_update_known_intermittent_2():179 tests = [("path/to/test.htm", [test_id], "testharness",180 b"""[test.htm]181 [test1]182 expected: PASS""")]183 log_0 = suite_log([("test_start", {"test": test_id}),184 ("test_status", {"test": test_id,185 "subtest": "test1",186 "status": "FAIL",187 "expected": "PASS"}),188 ("test_end", {"test": test_id,189 "status": "OK"})])190 updated = update(tests, log_0, update_intermittent=True)191 new_manifest = updated[0][1]192 assert not new_manifest.is_empty193 assert new_manifest.get_test(test_id).children[0].get(194 "expected", default_run_info) == "FAIL"195def test_update_existing_known_intermittent():196 tests = [("path/to/test.htm", [test_id], "testharness",197 b"""[test.htm]198 [test1]199 expected: [PASS, FAIL]""")]200 log_0 = suite_log([("test_start", {"test": test_id}),201 ("test_status", {"test": test_id,202 "subtest": "test1",203 "status": "ERROR",204 "expected": "PASS",205 "known_intermittent": ["FAIL"]}),206 ("test_end", {"test": test_id,207 "status": "OK"})])208 log_1 = suite_log([("test_start", {"test": test_id}),209 ("test_status", {"test": test_id,210 "subtest": "test1",211 "status": "PASS",212 "expected": "PASS",213 "known_intermittent": ["FAIL"]}),214 ("test_end", {"test": test_id,215 "status": "OK"})])216 log_2 = suite_log([("test_start", {"test": test_id}),217 ("test_status", {"test": test_id,218 "subtest": "test1",219 "status": "PASS",220 "expected": "PASS",221 "known_intermittent": ["FAIL"]}),222 ("test_end", {"test": test_id,223 "status": "OK"})])224 updated = update(tests, log_0, log_1, log_2, update_intermittent=True)225 new_manifest = updated[0][1]226 assert not new_manifest.is_empty227 assert new_manifest.get_test(test_id).children[0].get(228 "expected", default_run_info) == ["PASS", "ERROR", "FAIL"]229def test_update_remove_previous_intermittent():230 tests = [("path/to/test.htm", [test_id], "testharness",231 b"""[test.htm]232 [test1]233 expected: [PASS, FAIL]""")]234 log_0 = suite_log([("test_start", {"test": test_id}),235 ("test_status", {"test": test_id,236 "subtest": "test1",237 "status": "ERROR",238 "expected": "PASS",239 "known_intermittent": ["FAIL"]}),240 ("test_end", {"test": test_id,241 "status": "OK"})])242 log_1 = suite_log([("test_start", {"test": test_id}),243 ("test_status", {"test": test_id,244 "subtest": "test1",245 "status": "PASS",246 "expected": "PASS",247 "known_intermittent": ["FAIL"]}),248 ("test_end", {"test": test_id,249 "status": "OK"})])250 log_2 = suite_log([("test_start", {"test": test_id}),251 ("test_status", {"test": test_id,252 "subtest": "test1",253 "status": "PASS",254 "expected": "PASS",255 "known_intermittent": ["FAIL"]}),256 ("test_end", {"test": test_id,257 "status": "OK"})])258 updated = update(tests,259 log_0,260 log_1,261 log_2,262 update_intermittent=True,263 remove_intermittent=True)264 new_manifest = updated[0][1]265 assert not new_manifest.is_empty266 assert new_manifest.get_test(test_id).children[0].get(267 "expected", default_run_info) == ["PASS", "ERROR"]268def test_update_new_test_with_intermittent():269 tests = [("path/to/test.htm", [test_id], "testharness", None)]270 log_0 = suite_log([("test_start", {"test": test_id}),271 ("test_status", {"test": test_id,272 "subtest": "test1",273 "status": "PASS",274 "expected": "PASS"}),275 ("test_end", {"test": test_id,276 "status": "OK"})])277 log_1 = suite_log([("test_start", {"test": test_id}),278 ("test_status", {"test": test_id,279 "subtest": "test1",280 "status": "PASS",281 "expected": "PASS"}),282 ("test_end", {"test": test_id,283 "status": "OK"})])284 log_2 = suite_log([("test_start", {"test": test_id}),285 ("test_status", {"test": test_id,286 "subtest": "test1",287 "status": "FAIL",288 "expected": "PASS"}),289 ("test_end", {"test": test_id,290 "status": "OK"})])291 updated = update(tests, log_0, log_1, log_2, update_intermittent=True)292 new_manifest = updated[0][1]293 assert not new_manifest.is_empty294 assert new_manifest.get_test("test.htm") is None295 assert len(new_manifest.get_test(test_id).children) == 1296 assert new_manifest.get_test(test_id).children[0].get(297 "expected", default_run_info) == ["PASS", "FAIL"]298def test_update_expected_tie_resolution():299 tests = [("path/to/test.htm", [test_id], "testharness", None)]300 log_0 = suite_log([("test_start", {"test": test_id}),301 ("test_status", {"test": test_id,302 "subtest": "test1",303 "status": "PASS",304 "expected": "PASS"}),305 ("test_end", {"test": test_id,306 "status": "OK"})])307 log_1 = suite_log([("test_start", {"test": test_id}),308 ("test_status", {"test": test_id,309 "subtest": "test1",310 "status": "FAIL",311 "expected": "PASS"}),312 ("test_end", {"test": test_id,313 "status": "OK"})])314 updated = update(tests, log_0, log_1, update_intermittent=True)315 new_manifest = updated[0][1]316 assert not new_manifest.is_empty317 assert new_manifest.get_test(test_id).children[0].get(318 "expected", default_run_info) == ["PASS", "FAIL"]319def test_update_reorder_expected():320 tests = [("path/to/test.htm", [test_id], "testharness",321 b"""[test.htm]322 [test1]323 expected: [PASS, FAIL]""")]324 log_0 = suite_log([("test_start", {"test": test_id}),325 ("test_status", {"test": test_id,326 "subtest": "test1",327 "status": "FAIL",328 "expected": "PASS",329 "known_intermittent": ["FAIL"]}),330 ("test_end", {"test": test_id,331 "status": "OK"})])332 log_1 = suite_log([("test_start", {"test": test_id}),333 ("test_status", {"test": test_id,334 "subtest": "test1",335 "status": "FAIL",336 "expected": "PASS",337 "known_intermittent": ["FAIL"]}),338 ("test_end", {"test": test_id,339 "status": "OK"})])340 log_2 = suite_log([("test_start", {"test": test_id}),341 ("test_status", {"test": test_id,342 "subtest": "test1",343 "status": "PASS",344 "expected": "PASS",345 "known_intermittent": ["FAIL"]}),346 ("test_end", {"test": test_id,347 "status": "OK"})])348 updated = update(tests, log_0, log_1, log_2, update_intermittent=True)349 new_manifest = updated[0][1]350 assert not new_manifest.is_empty351 assert new_manifest.get_test(test_id).children[0].get(352 "expected", default_run_info) == ["FAIL", "PASS"]353def test_update_and_preserve_unchanged_expected_intermittent():354 tests = [("path/to/test.htm", [test_id], "testharness", b"""355[test.htm]356 expected:357 if os == "android": [PASS, FAIL]358 FAIL""")]359 log_0 = suite_log([("test_start", {"test": test_id}),360 ("test_end", {"test": test_id,361 "status": "FAIL",362 "expected": "PASS",363 "known_intermittent": ["FAIL"]})],364 run_info={"os": "android"})365 log_1 = suite_log([("test_start", {"test": test_id}),366 ("test_end", {"test": test_id,367 "status": "PASS",368 "expected": "PASS",369 "known_intermittent": ["FAIL"]})],370 run_info={"os": "android"})371 log_2 = suite_log([("test_start", {"test": test_id}),372 ("test_end", {"test": test_id,373 "status": "PASS",374 "expected": "FAIL"})])375 updated = update(tests, log_0, log_1, log_2)376 new_manifest = updated[0][1]377 assert not new_manifest.is_empty378 run_info_1 = default_run_info.copy()379 run_info_1.update({"os": "android"})380 assert not new_manifest.is_empty381 assert new_manifest.get_test(test_id).get(382 "expected", run_info_1) == ["PASS", "FAIL"]383 assert new_manifest.get_test(test_id).get(384 "expected", default_run_info) == "PASS"385def test_update_test_with_intermittent_to_one_expected_status():386 tests = [("path/to/test.htm", [test_id], "testharness",387 b"""[test.htm]388 [test1]389 expected: [PASS, FAIL]""")]390 log_0 = suite_log([("test_start", {"test": test_id}),391 ("test_status", {"test": test_id,392 "subtest": "test1",393 "status": "ERROR",394 "expected": "PASS",395 "known_intermittent": ["FAIL"]}),396 ("test_end", {"test": test_id,397 "status": "OK"})])398 updated = update(tests, log_0)399 new_manifest = updated[0][1]400 assert not new_manifest.is_empty401 assert new_manifest.get_test(test_id).children[0].get(402 "expected", default_run_info) == "ERROR"403def test_update_intermittent_with_conditions():404 tests = [("path/to/test.htm", [test_id], "testharness", b"""405[test.htm]406 expected:407 if os == "android": [PASS, FAIL]""")]408 log_0 = suite_log([("test_start", {"test": test_id}),409 ("test_end", {"test": test_id,410 "status": "TIMEOUT",411 "expected": "PASS",412 "known_intermittent": ["FAIL"]})],413 run_info={"os": "android"})414 log_1 = suite_log([("test_start", {"test": test_id}),415 ("test_end", {"test": test_id,416 "status": "PASS",417 "expected": "PASS",418 "known_intermittent": ["FAIL"]})],419 run_info={"os": "android"})420 updated = update(tests, log_0, log_1, update_intermittent=True)421 new_manifest = updated[0][1]422 assert not new_manifest.is_empty423 run_info_1 = default_run_info.copy()424 run_info_1.update({"os": "android"})425 assert not new_manifest.is_empty426 assert new_manifest.get_test(test_id).get(427 "expected", run_info_1) == ["PASS", "TIMEOUT", "FAIL"]428def test_update_and_remove_intermittent_with_conditions():429 tests = [("path/to/test.htm", [test_id], "testharness", b"""430[test.htm]431 expected:432 if os == "android": [PASS, FAIL]""")]433 log_0 = suite_log([("test_start", {"test": test_id}),434 ("test_end", {"test": test_id,435 "status": "TIMEOUT",436 "expected": "PASS",437 "known_intermittent": ["FAIL"]})],438 run_info={"os": "android"})439 log_1 = suite_log([("test_start", {"test": test_id}),440 ("test_end", {"test": test_id,441 "status": "PASS",442 "expected": "PASS",443 "known_intermittent": ["FAIL"]})],444 run_info={"os": "android"})445 updated = update(tests, log_0, log_1, update_intermittent=True, remove_intermittent=True)446 new_manifest = updated[0][1]447 assert not new_manifest.is_empty448 run_info_1 = default_run_info.copy()449 run_info_1.update({"os": "android"})450 assert not new_manifest.is_empty451 assert new_manifest.get_test(test_id).get(452 "expected", run_info_1) == ["PASS", "TIMEOUT"]453def test_update_intermittent_full():454 tests = [("path/to/test.htm", [test_id], "testharness",455 b"""[test.htm]456 [test1]457 expected:458 if os == "mac": [FAIL, TIMEOUT]459 FAIL""")]460 log_0 = suite_log([("test_start", {"test": test_id}),461 ("test_status", {"test": test_id,462 "subtest": "test1",463 "status": "FAIL",464 "expected": "FAIL",465 "known_intermittent": ["TIMEOUT"]}),466 ("test_end", {"test": test_id,467 "status": "OK"})],468 run_info={"os": "mac"})469 log_1 = suite_log([("test_start", {"test": test_id}),470 ("test_status", {"test": test_id,471 "subtest": "test1",472 "status": "FAIL"}),473 ("test_end", {"test": test_id,474 "status": "OK"})])475 updated = update(tests, log_0, log_1, update_intermittent=True, full_update=True)476 new_manifest = updated[0][1]477 assert not new_manifest.is_empty478 run_info_1 = default_run_info.copy()479 run_info_1.update({"os": "mac"})480 assert new_manifest.get_test(test_id).children[0].get(481 "expected", run_info_1) == ["FAIL", "TIMEOUT"]482 assert new_manifest.get_test(test_id).children[0].get(483 "expected", default_run_info) == "FAIL"484def test_update_intermittent_full_remove():485 tests = [("path/to/test.htm", [test_id], "testharness",486 b"""[test.htm]487 [test1]488 expected:489 if os == "mac": [FAIL, TIMEOUT, PASS]490 FAIL""")]491 log_0 = suite_log([("test_start", {"test": test_id}),492 ("test_status", {"test": test_id,493 "subtest": "test1",494 "status": "FAIL",495 "expected": "FAIL",496 "known_intermittent": ["TIMEOUT", "PASS"]}),497 ("test_end", {"test": test_id,498 "status": "OK"})],499 run_info={"os": "mac"})500 log_1 = suite_log([("test_start", {"test": test_id}),501 ("test_status", {"test": test_id,502 "subtest": "test1",503 "status": "TIMEOUT",504 "expected": "FAIL",505 "known_intermittent": ["TIMEOUT", "PASS"]}),506 ("test_end", {"test": test_id,507 "status": "OK"})],508 run_info={"os": "mac"})509 log_2 = suite_log([("test_start", {"test": test_id}),510 ("test_status", {"test": test_id,511 "subtest": "test1",512 "status": "FAIL"}),513 ("test_end", {"test": test_id,514 "status": "OK"})])515 updated = update(tests, log_0, log_1, log_2, update_intermittent=True,516 full_update=True, remove_intermittent=True)517 new_manifest = updated[0][1]518 assert not new_manifest.is_empty519 run_info_1 = default_run_info.copy()520 run_info_1.update({"os": "mac"})521 assert new_manifest.get_test(test_id).children[0].get(522 "expected", run_info_1) == ["FAIL", "TIMEOUT"]523 assert new_manifest.get_test(test_id).children[0].get(524 "expected", default_run_info) == "FAIL"525def test_full_update():526 tests = [("path/to/test.htm", [test_id], "testharness",527 b"""[test.htm]528 [test1]529 expected:530 if os == "mac": [FAIL, TIMEOUT]531 FAIL""")]532 log_0 = suite_log([("test_start", {"test": test_id}),533 ("test_status", {"test": test_id,534 "subtest": "test1",535 "status": "FAIL",536 "expected": "FAIL",537 "known_intermittent": ["TIMEOUT"]}),538 ("test_end", {"test": test_id,539 "status": "OK"})],540 run_info={"os": "mac"})541 log_1 = suite_log([("test_start", {"test": test_id}),542 ("test_status", {"test": test_id,543 "subtest": "test1",544 "status": "FAIL"}),545 ("test_end", {"test": test_id,546 "status": "OK"})])547 updated = update(tests, log_0, log_1, full_update=True)548 new_manifest = updated[0][1]549 assert not new_manifest.is_empty550 run_info_1 = default_run_info.copy()551 run_info_1.update({"os": "mac"})552 assert new_manifest.get_test(test_id).children[0].get(553 "expected", run_info_1) == "FAIL"554 assert new_manifest.get_test(test_id).children[0].get(555 "expected", default_run_info) == "FAIL"556def test_full_orphan():557 tests = [("path/to/test.htm", [test_id], "testharness",558 b"""[test.htm]559 [test1]560 expected: FAIL561 [subsub test]562 expected: TIMEOUT563 [test2]564 expected: ERROR565""")]566 log_0 = suite_log([("test_start", {"test": test_id}),567 ("test_status", {"test": test_id,568 "subtest": "test1",569 "status": "FAIL",570 "expected": "FAIL"}),571 ("test_end", {"test": test_id,572 "status": "OK"})])573 updated = update(tests, log_0, full_update=True)574 new_manifest = updated[0][1]575 assert not new_manifest.is_empty576 assert len(new_manifest.get_test(test_id).children[0].children) == 0577 assert new_manifest.get_test(test_id).children[0].get(578 "expected", default_run_info) == "FAIL"579 assert len(new_manifest.get_test(test_id).children) == 1580def test_update_reorder_expected_full_conditions():581 tests = [("path/to/test.htm", [test_id], "testharness",582 b"""[test.htm]583 [test1]584 expected:585 if os == "mac": [FAIL, TIMEOUT]586 [FAIL, PASS]""")]587 log_0 = suite_log([("test_start", {"test": test_id}),588 ("test_status", {"test": test_id,589 "subtest": "test1",590 "status": "TIMEOUT",591 "expected": "FAIL",592 "known_intermittent": ["TIMEOUT"]}),593 ("test_end", {"test": test_id,594 "status": "OK"})],595 run_info={"os": "mac"})596 log_1 = suite_log([("test_start", {"test": test_id}),597 ("test_status", {"test": test_id,598 "subtest": "test1",599 "status": "TIMEOUT",600 "expected": "FAIL",601 "known_intermittent": ["TIMEOUT"]}),602 ("test_end", {"test": test_id,603 "status": "OK"})],604 run_info={"os": "mac"})605 log_2 = suite_log([("test_start", {"test": test_id}),606 ("test_status", {"test": test_id,607 "subtest": "test1",608 "status": "PASS",609 "expected": "FAIL",610 "known_intermittent": ["PASS"]}),611 ("test_end", {"test": test_id,612 "status": "OK"})])613 log_3 = suite_log([("test_start", {"test": test_id}),614 ("test_status", {"test": test_id,615 "subtest": "test1",616 "status": "PASS",617 "expected": "FAIL",618 "known_intermittent": ["PASS"]}),619 ("test_end", {"test": test_id,620 "status": "OK"})])621 updated = update(tests, log_0, log_1, log_2, log_3, update_intermittent=True, full_update=True)622 new_manifest = updated[0][1]623 assert not new_manifest.is_empty624 run_info_1 = default_run_info.copy()625 run_info_1.update({"os": "mac"})626 assert new_manifest.get_test(test_id).children[0].get(627 "expected", run_info_1) == ["TIMEOUT", "FAIL"]628 assert new_manifest.get_test(test_id).children[0].get(629 "expected", default_run_info) == ["PASS", "FAIL"]630def test_skip_0():631 tests = [("path/to/test.htm", [test_id], "testharness",632 b"""[test.htm]633 [test1]634 expected: FAIL""")]635 log = suite_log([("test_start", {"test": test_id}),636 ("test_status", {"test": test_id,637 "subtest": "test1",638 "status": "FAIL",639 "expected": "FAIL"}),640 ("test_end", {"test": test_id,641 "status": "OK"})])642 updated = update(tests, log)643 assert not updated644def test_new_subtest():645 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]646 [test1]647 expected: FAIL""")]648 log = suite_log([("test_start", {"test": test_id}),649 ("test_status", {"test": test_id,650 "subtest": "test1",651 "status": "FAIL",652 "expected": "FAIL"}),653 ("test_status", {"test": test_id,654 "subtest": "test2",655 "status": "FAIL",656 "expected": "PASS"}),657 ("test_end", {"test": test_id,658 "status": "OK"})])659 updated = update(tests, log)660 new_manifest = updated[0][1]661 assert not new_manifest.is_empty662 assert new_manifest.get_test(test_id).children[0].get("expected", default_run_info) == "FAIL"663 assert new_manifest.get_test(test_id).children[1].get("expected", default_run_info) == "FAIL"664def test_update_multiple_0():665 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]666 [test1]667 expected: FAIL""")]668 log_0 = suite_log([("test_start", {"test": test_id}),669 ("test_status", {"test": test_id,670 "subtest": "test1",671 "status": "FAIL",672 "expected": "FAIL"}),673 ("test_end", {"test": test_id,674 "status": "OK"})],675 run_info={"debug": False, "os": "osx"})676 log_1 = suite_log([("test_start", {"test": test_id}),677 ("test_status", {"test": test_id,678 "subtest": "test1",679 "status": "TIMEOUT",680 "expected": "FAIL"}),681 ("test_end", {"test": test_id,682 "status": "OK"})],683 run_info={"debug": False, "os": "linux"})684 updated = update(tests, log_0, log_1)685 new_manifest = updated[0][1]686 assert not new_manifest.is_empty687 run_info_1 = default_run_info.copy()688 run_info_1.update({"debug": False, "os": "osx"})689 run_info_2 = default_run_info.copy()690 run_info_2.update({"debug": False, "os": "linux"})691 assert new_manifest.get_test(test_id).children[0].get(692 "expected", run_info_1) == "FAIL"693 assert new_manifest.get_test(test_id).children[0].get(694 "expected", {"debug": False, "os": "linux"}) == "TIMEOUT"695def test_update_multiple_1():696 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]697 [test1]698 expected: FAIL""")]699 log_0 = suite_log([("test_start", {"test": test_id}),700 ("test_status", {"test": test_id,701 "subtest": "test1",702 "status": "FAIL",703 "expected": "FAIL"}),704 ("test_end", {"test": test_id,705 "status": "OK"})],706 run_info={"os": "osx"})707 log_1 = suite_log([("test_start", {"test": test_id}),708 ("test_status", {"test": test_id,709 "subtest": "test1",710 "status": "TIMEOUT",711 "expected": "FAIL"}),712 ("test_end", {"test": test_id,713 "status": "OK"})],714 run_info={"os": "linux"})715 updated = update(tests, log_0, log_1)716 new_manifest = updated[0][1]717 assert not new_manifest.is_empty718 run_info_1 = default_run_info.copy()719 run_info_1.update({"os": "osx"})720 run_info_2 = default_run_info.copy()721 run_info_2.update({"os": "linux"})722 run_info_3 = default_run_info.copy()723 run_info_3.update({"os": "win"})724 assert new_manifest.get_test(test_id).children[0].get(725 "expected", run_info_1) == "FAIL"726 assert new_manifest.get_test(test_id).children[0].get(727 "expected", run_info_2) == "TIMEOUT"728 assert new_manifest.get_test(test_id).children[0].get(729 "expected", run_info_3) == "FAIL"730def test_update_multiple_2():731 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]732 [test1]733 expected: FAIL""")]734 log_0 = suite_log([("test_start", {"test": test_id}),735 ("test_status", {"test": test_id,736 "subtest": "test1",737 "status": "FAIL",738 "expected": "FAIL"}),739 ("test_end", {"test": test_id,740 "status": "OK"})],741 run_info={"debug": False, "os": "osx"})742 log_1 = suite_log([("test_start", {"test": test_id}),743 ("test_status", {"test": test_id,744 "subtest": "test1",745 "status": "TIMEOUT",746 "expected": "FAIL"}),747 ("test_end", {"test": test_id,748 "status": "OK"})],749 run_info={"debug": True, "os": "osx"})750 updated = update(tests, log_0, log_1)751 new_manifest = updated[0][1]752 run_info_1 = default_run_info.copy()753 run_info_1.update({"debug": False, "os": "osx"})754 run_info_2 = default_run_info.copy()755 run_info_2.update({"debug": True, "os": "osx"})756 assert not new_manifest.is_empty757 assert new_manifest.get_test(test_id).children[0].get(758 "expected", run_info_1) == "FAIL"759 assert new_manifest.get_test(test_id).children[0].get(760 "expected", run_info_2) == "TIMEOUT"761def test_update_multiple_3():762 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]763 [test1]764 expected:765 if debug: FAIL766 if not debug and os == "osx": TIMEOUT""")]767 log_0 = suite_log([("test_start", {"test": test_id}),768 ("test_status", {"test": test_id,769 "subtest": "test1",770 "status": "FAIL",771 "expected": "FAIL"}),772 ("test_end", {"test": test_id,773 "status": "OK"})],774 run_info={"debug": False, "os": "osx"})775 log_1 = suite_log([("test_start", {"test": test_id}),776 ("test_status", {"test": test_id,777 "subtest": "test1",778 "status": "TIMEOUT",779 "expected": "FAIL"}),780 ("test_end", {"test": test_id,781 "status": "OK"})],782 run_info={"debug": True, "os": "osx"})783 updated = update(tests, log_0, log_1)784 new_manifest = updated[0][1]785 run_info_1 = default_run_info.copy()786 run_info_1.update({"debug": False, "os": "osx"})787 run_info_2 = default_run_info.copy()788 run_info_2.update({"debug": True, "os": "osx"})789 assert not new_manifest.is_empty790 assert new_manifest.get_test(test_id).children[0].get(791 "expected", run_info_1) == "FAIL"792 assert new_manifest.get_test(test_id).children[0].get(793 "expected", run_info_2) == "TIMEOUT"794def test_update_ignore_existing():795 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]796 [test1]797 expected:798 if debug: TIMEOUT799 if not debug and os == "osx": NOTRUN""")]800 log_0 = suite_log([("test_start", {"test": test_id}),801 ("test_status", {"test": test_id,802 "subtest": "test1",803 "status": "FAIL",804 "expected": "PASS"}),805 ("test_end", {"test": test_id,806 "status": "OK"})],807 run_info={"debug": False, "os": "linux"})808 log_1 = suite_log([("test_start", {"test": test_id}),809 ("test_status", {"test": test_id,810 "subtest": "test1",811 "status": "FAIL",812 "expected": "PASS"}),813 ("test_end", {"test": test_id,814 "status": "OK"})],815 run_info={"debug": True, "os": "windows"})816 updated = update(tests, log_0, log_1)817 new_manifest = updated[0][1]818 run_info_1 = default_run_info.copy()819 run_info_1.update({"debug": False, "os": "linux"})820 run_info_2 = default_run_info.copy()821 run_info_2.update({"debug": False, "os": "osx"})822 assert not new_manifest.is_empty823 assert new_manifest.get_test(test_id).children[0].get(824 "expected", run_info_1) == "FAIL"825 assert new_manifest.get_test(test_id).children[0].get(826 "expected", run_info_2) == "NOTRUN"827def test_update_new_test():828 tests = [("path/to/test.htm", [test_id], "testharness", None)]829 log_0 = suite_log([("test_start", {"test": test_id}),830 ("test_status", {"test": test_id,831 "subtest": "test1",832 "status": "FAIL",833 "expected": "PASS"}),834 ("test_end", {"test": test_id,835 "status": "OK"})])836 updated = update(tests, log_0)837 new_manifest = updated[0][1]838 run_info_1 = default_run_info.copy()839 assert not new_manifest.is_empty840 assert new_manifest.get_test("test.htm") is None841 assert len(new_manifest.get_test(test_id).children) == 1842 assert new_manifest.get_test(test_id).children[0].get(843 "expected", run_info_1) == "FAIL"844def test_update_duplicate():845 tests = [("path/to/test.htm", [test_id], "testharness", b"""846[test.htm]847 expected: ERROR""")]848 log_0 = suite_log([("test_start", {"test": test_id}),849 ("test_end", {"test": test_id,850 "status": "PASS"})])851 log_1 = suite_log([("test_start", {"test": test_id}),852 ("test_end", {"test": test_id,853 "status": "FAIL"})])854 updated = update(tests, log_0, log_1)855 new_manifest = updated[0][1]856 run_info_1 = default_run_info.copy()857 assert new_manifest.get_test(test_id).get(858 "expected", run_info_1) == "ERROR"859def test_update_disable_intermittent():860 tests = [("path/to/test.htm", [test_id], "testharness", b"""861[test.htm]862 expected: ERROR""")]863 log_0 = suite_log([("test_start", {"test": test_id}),864 ("test_end", {"test": test_id,865 "status": "PASS"})])866 log_1 = suite_log([("test_start", {"test": test_id}),867 ("test_end", {"test": test_id,868 "status": "FAIL"})])869 updated = update(tests, log_0, log_1, disable_intermittent="Some message")870 new_manifest = updated[0][1]871 run_info_1 = default_run_info.copy()872 assert new_manifest.get_test(test_id).get(873 "disabled", run_info_1) == "Some message"874def test_update_stability_conditional_instability():875 tests = [("path/to/test.htm", [test_id], "testharness", b"""876[test.htm]877 expected: ERROR""")]878 log_0 = suite_log([("test_start", {"test": test_id}),879 ("test_end", {"test": test_id,880 "status": "PASS"})],881 run_info={"os": "linux"})882 log_1 = suite_log([("test_start", {"test": test_id}),883 ("test_end", {"test": test_id,884 "status": "FAIL"})],885 run_info={"os": "linux"})886 log_2 = suite_log([("test_start", {"test": test_id}),887 ("test_end", {"test": test_id,888 "status": "FAIL"})],889 run_info={"os": "mac"})890 updated = update(tests, log_0, log_1, log_2, disable_intermittent="Some message")891 new_manifest = updated[0][1]892 run_info_1 = default_run_info.copy()893 run_info_1.update({"os": "linux"})894 run_info_2 = default_run_info.copy()895 run_info_2.update({"os": "mac"})896 assert new_manifest.get_test(test_id).get(897 "disabled", run_info_1) == "Some message"898 with pytest.raises(KeyError):899 assert new_manifest.get_test(test_id).get(900 "disabled", run_info_2)901 assert new_manifest.get_test(test_id).get(902 "expected", run_info_2) == "FAIL"903def test_update_full():904 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]905 [test1]906 expected:907 if debug: TIMEOUT908 if not debug and os == "osx": NOTRUN909 [test2]910 expected: FAIL911[test.js]912 [test1]913 expected: FAIL914""")]915 log_0 = suite_log([("test_start", {"test": test_id}),916 ("test_status", {"test": test_id,917 "subtest": "test1",918 "status": "FAIL",919 "expected": "PASS"}),920 ("test_end", {"test": test_id,921 "status": "OK"})],922 run_info={"debug": False})923 log_1 = suite_log([("test_start", {"test": test_id}),924 ("test_status", {"test": test_id,925 "subtest": "test1",926 "status": "ERROR",927 "expected": "PASS"}),928 ("test_end", {"test": test_id,929 "status": "OK"})],930 run_info={"debug": True})931 updated = update(tests, log_0, log_1, full_update=True)932 new_manifest = updated[0][1]933 run_info_1 = default_run_info.copy()934 run_info_1.update({"debug": False, "os": "win"})935 run_info_2 = default_run_info.copy()936 run_info_2.update({"debug": True, "os": "osx"})937 assert not new_manifest.is_empty938 assert new_manifest.get_test("test.js") is None939 assert len(new_manifest.get_test(test_id).children) == 1940 assert new_manifest.get_test(test_id).children[0].get(941 "expected", run_info_1) == "FAIL"942 assert new_manifest.get_test(test_id).children[0].get(943 "expected", run_info_2) == "ERROR"944def test_update_full_unknown():945 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]946 [test1]947 expected:948 if release_or_beta: ERROR949 if not debug and os == "osx": NOTRUN950""")]951 log_0 = suite_log([("test_start", {"test": test_id}),952 ("test_status", {"test": test_id,953 "subtest": "test1",954 "status": "FAIL",955 "expected": "PASS"}),956 ("test_end", {"test": test_id,957 "status": "OK"})],958 run_info={"debug": False, "release_or_beta": False})959 log_1 = suite_log([("test_start", {"test": test_id}),960 ("test_status", {"test": test_id,961 "subtest": "test1",962 "status": "FAIL",963 "expected": "PASS"}),964 ("test_end", {"test": test_id,965 "status": "OK"})],966 run_info={"debug": True, "release_or_beta": False})967 updated = update(tests, log_0, log_1, full_update=True)968 new_manifest = updated[0][1]969 run_info_1 = default_run_info.copy()970 run_info_1.update({"release_or_beta": False})971 run_info_2 = default_run_info.copy()972 run_info_2.update({"release_or_beta": True})973 assert not new_manifest.is_empty974 assert new_manifest.get_test(test_id).children[0].get(975 "expected", run_info_1) == "FAIL"976 assert new_manifest.get_test(test_id).children[0].get(977 "expected", run_info_2) == "ERROR"978def test_update_full_unknown_missing():979 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]980 [subtest_deleted]981 expected:982 if release_or_beta: ERROR983 FAIL984""")]985 log_0 = suite_log([("test_start", {"test": test_id}),986 ("test_status", {"test": test_id,987 "subtest": "test1",988 "status": "PASS",989 "expected": "PASS"}),990 ("test_end", {"test": test_id,991 "status": "OK"})],992 run_info={"debug": False, "release_or_beta": False})993 updated = update(tests, log_0, full_update=True)994 assert len(updated) == 0995def test_update_default():996 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]997 [test1]998 expected:999 if os == "mac": FAIL1000 ERROR""")]1001 log_0 = suite_log([("test_start", {"test": test_id}),1002 ("test_status", {"test": test_id,1003 "subtest": "test1",1004 "status": "PASS",1005 "expected": "FAIL"}),1006 ("test_end", {"test": test_id,1007 "status": "OK"})],1008 run_info={"os": "mac"})1009 log_1 = suite_log([("test_start", {"test": test_id}),1010 ("test_status", {"test": test_id,1011 "subtest": "test1",1012 "status": "PASS",1013 "expected": "ERROR"}),1014 ("test_end", {"test": test_id,1015 "status": "OK"})],1016 run_info={"os": "linux"})1017 updated = update(tests, log_0, log_1)1018 new_manifest = updated[0][1]1019 assert new_manifest.is_empty1020def test_update_default_1():1021 tests = [("path/to/test.htm", [test_id], "testharness", b"""1022[test.htm]1023 expected:1024 if os == "mac": TIMEOUT1025 ERROR""")]1026 log_0 = suite_log([("test_start", {"test": test_id}),1027 ("test_end", {"test": test_id,1028 "expected": "ERROR",1029 "status": "FAIL"})],1030 run_info={"os": "linux"})1031 updated = update(tests, log_0)1032 new_manifest = updated[0][1]1033 assert not new_manifest.is_empty1034 run_info_1 = default_run_info.copy()1035 run_info_1.update({"os": "mac"})1036 run_info_2 = default_run_info.copy()1037 run_info_2.update({"os": "win"})1038 assert not new_manifest.is_empty1039 assert new_manifest.get_test(test_id).get(1040 "expected", run_info_1) == "TIMEOUT"1041 assert new_manifest.get_test(test_id).get(1042 "expected", run_info_2) == "FAIL"1043def test_update_default_2():1044 tests = [("path/to/test.htm", [test_id], "testharness", b"""1045[test.htm]1046 expected:1047 if os == "mac": TIMEOUT1048 ERROR""")]1049 log_0 = suite_log([("test_start", {"test": test_id}),1050 ("test_end", {"test": test_id,1051 "expected": "ERROR",1052 "status": "TIMEOUT"})],1053 run_info={"os": "linux"})1054 updated = update(tests, log_0)1055 new_manifest = updated[0][1]1056 assert not new_manifest.is_empty1057 run_info_1 = default_run_info.copy()1058 run_info_1.update({"os": "mac"})1059 run_info_2 = default_run_info.copy()1060 run_info_2.update({"os": "win"})1061 assert not new_manifest.is_empty1062 assert new_manifest.get_test(test_id).get(1063 "expected", run_info_1) == "TIMEOUT"1064 assert new_manifest.get_test(test_id).get(1065 "expected", run_info_2) == "TIMEOUT"1066def test_update_assertion_count_0():1067 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]1068 max-asserts: 41069 min-asserts: 21070""")]1071 log_0 = suite_log([("test_start", {"test": test_id}),1072 ("assertion_count", {"test": test_id,1073 "count": 6,1074 "min_expected": 2,1075 "max_expected": 4}),1076 ("test_end", {"test": test_id,1077 "status": "OK"})])1078 updated = update(tests, log_0)1079 new_manifest = updated[0][1]1080 assert not new_manifest.is_empty1081 assert new_manifest.get_test(test_id).get("max-asserts") == "7"1082 assert new_manifest.get_test(test_id).get("min-asserts") == "2"1083def test_update_assertion_count_1():1084 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]1085 max-asserts: 41086 min-asserts: 21087""")]1088 log_0 = suite_log([("test_start", {"test": test_id}),1089 ("assertion_count", {"test": test_id,1090 "count": 1,1091 "min_expected": 2,1092 "max_expected": 4}),1093 ("test_end", {"test": test_id,1094 "status": "OK"})])1095 updated = update(tests, log_0)1096 new_manifest = updated[0][1]1097 assert not new_manifest.is_empty1098 assert new_manifest.get_test(test_id).get("max-asserts") == "4"1099 assert new_manifest.get_test(test_id).has_key("min-asserts") is False1100def test_update_assertion_count_2():1101 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]1102 max-asserts: 41103 min-asserts: 21104""")]1105 log_0 = suite_log([("test_start", {"test": test_id}),1106 ("assertion_count", {"test": test_id,1107 "count": 3,1108 "min_expected": 2,1109 "max_expected": 4}),1110 ("test_end", {"test": test_id,1111 "status": "OK"})])1112 updated = update(tests, log_0)1113 assert not updated1114def test_update_assertion_count_3():1115 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]1116 max-asserts: 41117 min-asserts: 21118""")]1119 log_0 = suite_log([("test_start", {"test": test_id}),1120 ("assertion_count", {"test": test_id,1121 "count": 6,1122 "min_expected": 2,1123 "max_expected": 4}),1124 ("test_end", {"test": test_id,1125 "status": "OK"})],1126 run_info={"os": "windows"})1127 log_1 = suite_log([("test_start", {"test": test_id}),1128 ("assertion_count", {"test": test_id,1129 "count": 7,1130 "min_expected": 2,1131 "max_expected": 4}),1132 ("test_end", {"test": test_id,1133 "status": "OK"})],1134 run_info={"os": "linux"})1135 updated = update(tests, log_0, log_1)1136 new_manifest = updated[0][1]1137 assert not new_manifest.is_empty1138 assert new_manifest.get_test(test_id).get("max-asserts") == "8"1139 assert new_manifest.get_test(test_id).get("min-asserts") == "2"1140def test_update_assertion_count_4():1141 tests = [("path/to/test.htm", [test_id], "testharness", b"""[test.htm]""")]1142 log_0 = suite_log([("test_start", {"test": test_id}),1143 ("assertion_count", {"test": test_id,1144 "count": 6,1145 "min_expected": 0,1146 "max_expected": 0}),1147 ("test_end", {"test": test_id,1148 "status": "OK"})],1149 run_info={"os": "windows"})1150 log_1 = suite_log([("test_start", {"test": test_id}),1151 ("assertion_count", {"test": test_id,1152 "count": 7,1153 "min_expected": 0,1154 "max_expected": 0}),1155 ("test_end", {"test": test_id,1156 "status": "OK"})],1157 run_info={"os": "linux"})1158 updated = update(tests, log_0, log_1)1159 new_manifest = updated[0][1]1160 assert not new_manifest.is_empty1161 assert new_manifest.get_test(test_id).get("max-asserts") == "8"1162 assert new_manifest.get_test(test_id).has_key("min-asserts") is False1163def test_update_lsan_0():1164 tests = [("path/to/test.htm", [test_id], "testharness", b""),1165 ("path/to/__dir__", [dir_id], None, b"")]1166 log_0 = suite_log([("lsan_leak", {"scope": "path/to/",1167 "frames": ["foo", "bar"]})])1168 updated = update(tests, log_0)1169 new_manifest = updated[0][1]1170 assert not new_manifest.is_empty1171 assert new_manifest.get("lsan-allowed") == ["foo"]1172def test_update_lsan_1():1173 tests = [("path/to/test.htm", [test_id], "testharness", b""),1174 ("path/to/__dir__", [dir_id], None, b"""1175lsan-allowed: [foo]""")]1176 log_0 = suite_log([("lsan_leak", {"scope": "path/to/",1177 "frames": ["foo", "bar"]}),1178 ("lsan_leak", {"scope": "path/to/",1179 "frames": ["baz", "foobar"]})])1180 updated = update(tests, log_0)1181 new_manifest = updated[0][1]1182 assert not new_manifest.is_empty1183 assert new_manifest.get("lsan-allowed") == ["baz", "foo"]1184def test_update_lsan_2():1185 tests = [("path/to/test.htm", [test_id], "testharness", b""),1186 ("path/__dir__", ["path/__dir__"], None, b"""1187lsan-allowed: [foo]"""),1188 ("path/to/__dir__", [dir_id], None, b"")]1189 log_0 = suite_log([("lsan_leak", {"scope": "path/to/",1190 "frames": ["foo", "bar"],1191 "allowed_match": ["foo"]}),1192 ("lsan_leak", {"scope": "path/to/",1193 "frames": ["baz", "foobar"]})])1194 updated = update(tests, log_0)1195 new_manifest = updated[0][1]1196 assert not new_manifest.is_empty1197 assert new_manifest.get("lsan-allowed") == ["baz"]1198def test_update_lsan_3():1199 tests = [("path/to/test.htm", [test_id], "testharness", b""),1200 ("path/to/__dir__", [dir_id], None, b"")]1201 log_0 = suite_log([("lsan_leak", {"scope": "path/to/",1202 "frames": ["foo", "bar"]})],1203 run_info={"os": "win"})1204 log_1 = suite_log([("lsan_leak", {"scope": "path/to/",1205 "frames": ["baz", "foobar"]})],1206 run_info={"os": "linux"})1207 updated = update(tests, log_0, log_1)1208 new_manifest = updated[0][1]1209 assert not new_manifest.is_empty1210 assert new_manifest.get("lsan-allowed") == ["baz", "foo"]1211def test_update_wptreport_0():1212 tests = [("path/to/test.htm", [test_id], "testharness",1213 b"""[test.htm]1214 [test1]1215 expected: FAIL""")]1216 log = {"run_info": default_run_info.copy(),1217 "results": [1218 {"test": "/path/to/test.htm",1219 "subtests": [{"name": "test1",1220 "status": "PASS",1221 "expected": "FAIL"}],1222 "status": "OK"}]}1223 updated = update(tests, log)1224 assert len(updated) == 11225 assert updated[0][1].is_empty1226def test_update_wptreport_1():1227 tests = [("path/to/test.htm", [test_id], "testharness", b""),1228 ("path/to/__dir__", [dir_id], None, b"")]1229 log = {"run_info": default_run_info.copy(),1230 "results": [],1231 "lsan_leaks": [{"scope": "path/to/",1232 "frames": ["baz", "foobar"]}]}1233 updated = update(tests, log)1234 assert len(updated) == 11235 assert updated[0][1].get("lsan-allowed") == ["baz"]1236def test_update_leak_total_0():1237 tests = [("path/to/test.htm", [test_id], "testharness", b""),1238 ("path/to/__dir__", [dir_id], None, b"")]1239 log_0 = suite_log([("mozleak_total", {"scope": "path/to/",1240 "process": "default",1241 "bytes": 100,1242 "threshold": 0,1243 "objects": []})])1244 updated = update(tests, log_0)1245 new_manifest = updated[0][1]1246 assert not new_manifest.is_empty1247 assert new_manifest.get("leak-threshold") == ['default:51200']1248def test_update_leak_total_1():1249 tests = [("path/to/test.htm", [test_id], "testharness", b""),1250 ("path/to/__dir__", [dir_id], None, b"")]1251 log_0 = suite_log([("mozleak_total", {"scope": "path/to/",1252 "process": "default",1253 "bytes": 100,1254 "threshold": 1000,1255 "objects": []})])1256 updated = update(tests, log_0)1257 assert not updated1258def test_update_leak_total_2():1259 tests = [("path/to/test.htm", [test_id], "testharness", b""),1260 ("path/to/__dir__", [dir_id], None, b"""1261leak-total: 110""")]1262 log_0 = suite_log([("mozleak_total", {"scope": "path/to/",1263 "process": "default",1264 "bytes": 100,1265 "threshold": 110,1266 "objects": []})])1267 updated = update(tests, log_0)1268 assert not updated1269def test_update_leak_total_3():1270 tests = [("path/to/test.htm", [test_id], "testharness", b""),1271 ("path/to/__dir__", [dir_id], None, b"""1272leak-total: 100""")]1273 log_0 = suite_log([("mozleak_total", {"scope": "path/to/",1274 "process": "default",1275 "bytes": 1000,1276 "threshold": 100,1277 "objects": []})])1278 updated = update(tests, log_0)1279 new_manifest = updated[0][1]1280 assert not new_manifest.is_empty1281 assert new_manifest.get("leak-threshold") == ['default:51200']1282def test_update_leak_total_4():1283 tests = [("path/to/test.htm", [test_id], "testharness", b""),1284 ("path/to/__dir__", [dir_id], None, b"""1285leak-total: 110""")]1286 log_0 = suite_log([1287 ("lsan_leak", {"scope": "path/to/",1288 "frames": ["foo", "bar"]}),1289 ("mozleak_total", {"scope": "path/to/",1290 "process": "default",1291 "bytes": 100,1292 "threshold": 110,1293 "objects": []})])1294 updated = update(tests, log_0)1295 new_manifest = updated[0][1]1296 assert not new_manifest.is_empty1297 assert new_manifest.has_key("leak-threshold") is False1298class TestStep(Step):1299 def create(self, state):1300 tests = [("path/to/test.htm", [test_id], "testharness", "")]1301 state.foo = create_test_manifest(tests)1302class UpdateRunner(StepRunner):1303 steps = [TestStep]1304def test_update_pickle():1305 logger = structuredlog.StructuredLogger("expected_test")1306 args = {1307 "test_paths": {1308 "/": {"tests_path": os.path.abspath(os.path.join(here,1309 os.pardir,1310 os.pardir,1311 os.pardir,1312 os.pardir))},1313 },1314 "abort": False,1315 "continue": False,1316 "sync": False,1317 }1318 args2 = args.copy()1319 args2["abort"] = True1320 wptupdate = WPTUpdate(logger, **args2)1321 wptupdate = WPTUpdate(logger, runner_cls=UpdateRunner, **args)...

Full Screen

Full Screen

stock_analysis3.py

Source:stock_analysis3.py Github

copy

Full Screen

1import csv2from matplotlib import pyplot as plt3from datetime import datetime4"""5不同下跌天数买入,第二天均以收盘价卖出策略,它和m>1.1时stock_analysis2的结果是一样的6"""7def get_data_analysis3(stock_code, stock_name, test_start, test_end, m):8 global dates, open_prices, highs, lows, closes, precloses, pctChgs9 filename = '/home/cdd/Desktop/Scraping/stocks/stock_analysis/stock_data/' + stock_code + stock_name + '.csv'10 with open(filename) as f:11 reader = csv.reader(f)12 dates, open_prices, highs, lows, closes, precloses, pctChgs = [], [], [], [], [], [], []13 for row in reader:14 try:15 current_date = datetime.strptime(row[0], '%Y-%m-%d')16 open_price = float(row[1])17 high = float(row[2])18 low = float(row[3])19 close = float(row[4])20 preclose = float(row[5])21 pctChg = float(row[6])22 except ValueError:23 pass24 else:25 dates.append(current_date)26 open_prices.append(open_price)27 highs.append(high)28 lows.append(low)29 closes.append(close)30 precloses.append(preclose)31 pctChgs.append(pctChg)32 """首跌买入策略"""33 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金34 stock_nums, stock_costs, account_profits_1 = [0], [], [] # 账户股票数量、股票成本、账户收益35 acts = []36 for i in range(test_start, test_end):37 if stock_nums[-1] == 0 and pctChgs[i-1] >= 0 and pctChgs[i] < 0: # 账户空仓、当日下跌买入38 act = '买入'39 stock_num = account_sums[-1] // closes[i] # 买入股票数量40 account_stock = stock_num * closes[i] # 股票市值41 account_cash = account_sums[-1] - account_stock # 现金余额42 account_sum = account_cash + account_stock # 当日账户总额43 stock_cost = closes[i] # 股票成本44 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益45 elif stock_nums[-1] != 0:46 act = '卖出'47 """以收盘价卖出"""48 stock_num = 0 # 清仓,股票数量为049 account_stock = 0 # 股票市值为050 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金51 account_sum = account_cash + account_stock # 当日账户总额52 stock_cost = 053 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益54 else:55 act = '空仓'56 stock_num = 0 # 空仓,股票数量为057 account_stock = 0 # 空仓,股票市值为058 account_cash = account_cashes[-1] # 账户现金不变59 account_sum = account_sums[-1] # 账户总额不变60 stock_cost = 061 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益62 """更新账户"""63 account_sums.append(account_sum)64 account_cashes.append(account_cash)65 account_stocks.append(account_stock)66 stock_nums.append(stock_num)67 stock_costs.append(stock_cost)68 account_profits_1.append(account_profit)69 acts.append(act)70 """绘制收益曲线"""71 # fig = plt.figure(dpi=128, figsize=(10, 6))72 # plt.plot(dates[test_start:test_end], account_profits_1, c='red')73 # plt.title(stock_code + '(n=1, m=' + str(m) + ')')74 # fig.autofmt_xdate()75 # plt.show()76 """连续两日下跌买入策略"""77 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金78 stock_nums, stock_costs, account_profits_2 = [0], [], [] # 账户股票数量、股票成本、账户收益79 acts = []80 for i in range(test_start, test_end):81 if stock_nums[-1] == 0 and pctChgs[i-2] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0: # 账户空仓、当日下跌买入82 act = '买入'83 stock_num = account_sums[-1] // closes[i] # 买入股票数量84 account_stock = stock_num * closes[i] # 股票市值85 account_cash = account_sums[-1] - account_stock # 现金余额86 account_sum = account_cash + account_stock # 当日账户总额87 stock_cost = closes[i] # 股票成本88 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益89 elif stock_nums[-1] != 0:90 act = '卖出'91 """以收盘价卖出"""92 stock_num = 0 # 清仓,股票数量为093 account_stock = 0 # 股票市值为094 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金95 account_sum = account_cash + account_stock # 当日账户总额96 stock_cost = 097 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益98 else:99 act = '空仓'100 stock_num = 0 # 空仓,股票数量为0101 account_stock = 0 # 空仓,股票市值为0102 account_cash = account_cashes[-1] # 账户现金不变103 account_sum = account_sums[-1] # 账户总额不变104 stock_cost = 0105 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益106 """更新账户"""107 account_sums.append(account_sum)108 account_cashes.append(account_cash)109 account_stocks.append(account_stock)110 stock_nums.append(stock_num)111 stock_costs.append(stock_cost)112 account_profits_2.append(account_profit)113 acts.append(act)114 """绘制收益曲线"""115 # fig = plt.figure(dpi=128, figsize=(10, 6))116 # plt.plot(dates[test_start:test_end], account_profits_2, c='red')117 # plt.title(stock_code + '(n=2)')118 # fig.autofmt_xdate()119 # plt.show()120 """连续三日下跌买入策略"""121 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金122 stock_nums, stock_costs, account_profits_3 = [0], [], [] # 账户股票数量、股票成本、账户收益123 acts = []124 for i in range(test_start, test_end):125 if stock_nums[-1] == 0 and pctChgs[i-3] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0 and pctChgs[i-2] < 0: # 账户空仓、当日下跌买入126 act = '买入'127 stock_num = account_sums[-1] // closes[i] # 买入股票数量128 account_stock = stock_num * closes[i] # 股票市值129 account_cash = account_sums[-1] - account_stock # 现金余额130 account_sum = account_cash + account_stock # 当日账户总额131 stock_cost = closes[i] # 股票成本132 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益133 elif stock_nums[-1] != 0:134 act = '卖出'135 """以收盘价卖出"""136 stock_num = 0 # 清仓,股票数量为0137 account_stock = 0 # 股票市值为0138 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金139 account_sum = account_cash + account_stock # 当日账户总额140 stock_cost = 0141 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益142 else:143 act = '空仓'144 stock_num = 0 # 空仓,股票数量为0145 account_stock = 0 # 空仓,股票市值为0146 account_cash = account_cashes[-1] # 账户现金不变147 account_sum = account_sums[-1] # 账户总额不变148 stock_cost = 0149 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益150 """更新账户"""151 account_sums.append(account_sum)152 account_cashes.append(account_cash)153 account_stocks.append(account_stock)154 stock_nums.append(stock_num)155 stock_costs.append(stock_cost)156 account_profits_3.append(account_profit)157 acts.append(act)158 """绘制收益曲线"""159 # fig = plt.figure(dpi=128, figsize=(10, 6))160 # plt.plot(dates[test_start:test_end], account_profits_3, c='red')161 # plt.title(stock_code + '(n=3)')162 # fig.autofmt_xdate()163 # plt.show()164 """165 连续四日下跌买入策略166 """167 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金168 stock_nums, stock_costs, account_profits_4 = [0], [], [] # 账户股票数量、股票成本、账户收益169 acts = []170 for i in range(test_start, test_end):171 if stock_nums[-1] == 0 and pctChgs[i-4] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0 and pctChgs[i-2] < 0 and pctChgs[i-3] < 0: # 账户空仓、当日下跌买入172 act = '买入'173 stock_num = account_sums[-1] // closes[i] # 买入股票数量174 account_stock = stock_num * closes[i] # 股票市值175 account_cash = account_sums[-1] - account_stock # 现金余额176 account_sum = account_cash + account_stock # 当日账户总额177 stock_cost = closes[i] # 股票成本178 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益179 elif stock_nums[-1] != 0:180 act = '卖出'181 """以收盘价卖出"""182 stock_num = 0 # 清仓,股票数量为0183 account_stock = 0 # 股票市值为0184 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金185 account_sum = account_cash + account_stock # 当日账户总额186 stock_cost = 0187 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益188 else:189 act = '空仓'190 stock_num = 0 # 空仓,股票数量为0191 account_stock = 0 # 空仓,股票市值为0192 account_cash = account_cashes[-1] # 账户现金不变193 account_sum = account_sums[-1] # 账户总额不变194 stock_cost = 0195 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益196 """更新账户"""197 account_sums.append(account_sum)198 account_cashes.append(account_cash)199 account_stocks.append(account_stock)200 stock_nums.append(stock_num)201 stock_costs.append(stock_cost)202 account_profits_4.append(account_profit)203 acts.append(act)204 """绘制收益曲线"""205 # fig = plt.figure(dpi=128, figsize=(10, 6))206 # plt.plot(dates[test_start:test_end], account_profits_4, c='red')207 # plt.title(stock_code + '(n=4)')208 # fig.autofmt_xdate()209 # plt.show()210 """211 连续五日下跌买入策略212 """213 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金214 stock_nums, stock_costs, account_profits_5 = [0], [], [] # 账户股票数量、股票成本、账户收益215 acts = []216 for i in range(test_start, test_end):217 if stock_nums[-1] == 0 and pctChgs[i-5] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0 and pctChgs[i-2] < 0 \218 and pctChgs[i-3] < 0 and pctChgs[i-4] < 0: # 账户空仓、当日下跌买入219 act = '买入'220 stock_num = account_sums[-1] // closes[i] # 买入股票数量221 account_stock = stock_num * closes[i] # 股票市值222 account_cash = account_sums[-1] - account_stock # 现金余额223 account_sum = account_cash + account_stock # 当日账户总额224 stock_cost = closes[i] # 股票成本225 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益226 elif stock_nums[-1] != 0:227 act = '卖出'228 """以收盘价卖出"""229 stock_num = 0 # 清仓,股票数量为0230 account_stock = 0 # 股票市值为0231 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金232 account_sum = account_cash + account_stock # 当日账户总额233 stock_cost = 0234 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益235 else:236 act = '空仓'237 stock_num = 0 # 空仓,股票数量为0238 account_stock = 0 # 空仓,股票市值为0239 account_cash = account_cashes[-1] # 账户现金不变240 account_sum = account_sums[-1] # 账户总额不变241 stock_cost = 0242 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益243 """更新账户"""244 account_sums.append(account_sum)245 account_cashes.append(account_cash)246 account_stocks.append(account_stock)247 stock_nums.append(stock_num)248 stock_costs.append(stock_cost)249 account_profits_5.append(account_profit)250 acts.append(act)251 """绘制收益曲线"""252 # fig = plt.figure(dpi=128, figsize=(10, 6))253 # plt.plot(dates[test_start:test_end], account_profits_5, c='red')254 # plt.title(stock_code + '(n=5)')255 # fig.autofmt_xdate()256 # plt.show()257 """258 连续六日下跌买入策略259 """260 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金261 stock_nums, stock_costs, account_profits_6 = [0], [], [] # 账户股票数量、股票成本、账户收益262 acts = []263 for i in range(test_start, test_end):264 if stock_nums[-1] == 0 and pctChgs[i-6] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0 and pctChgs[i-2] < 0 \265 and pctChgs[i-3] < 0 and pctChgs[i-4] < 0 and pctChgs[i-5] < 0: # 账户空仓、当日下跌买入266 act = '买入'267 stock_num = account_sums[-1] // closes[i] # 买入股票数量268 account_stock = stock_num * closes[i] # 股票市值269 account_cash = account_sums[-1] - account_stock # 现金余额270 account_sum = account_cash + account_stock # 当日账户总额271 stock_cost = closes[i] # 股票成本272 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益273 elif stock_nums[-1] != 0:274 act = '卖出'275 """以收盘价卖出"""276 stock_num = 0 # 清仓,股票数量为0277 account_stock = 0 # 股票市值为0278 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金279 account_sum = account_cash + account_stock # 当日账户总额280 stock_cost = 0281 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益282 else:283 act = '空仓'284 stock_num = 0 # 空仓,股票数量为0285 account_stock = 0 # 空仓,股票市值为0286 account_cash = account_cashes[-1] # 账户现金不变287 account_sum = account_sums[-1] # 账户总额不变288 stock_cost = 0289 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益290 """更新账户"""291 account_sums.append(account_sum)292 account_cashes.append(account_cash)293 account_stocks.append(account_stock)294 stock_nums.append(stock_num)295 stock_costs.append(stock_cost)296 account_profits_6.append(account_profit)297 acts.append(act)298 """绘制收益曲线"""299 # fig = plt.figure(dpi=128, figsize=(10, 6))300 # plt.plot(dates[test_start:test_end], account_profits_6, c='red')301 # plt.title(stock_code + '(n=6)')302 # fig.autofmt_xdate()303 # plt.show()304 """305 连续七日下跌买入策略306 """307 account_sums, account_stocks, account_cashes = [1000000], [0], [1000000] # 总账户金额、股票市值、账户现金308 stock_nums, stock_costs, account_profits_7 = [0], [], [] # 账户股票数量、股票成本、账户收益309 acts = []310 for i in range(test_start, test_end):311 if stock_nums[-1] == 0 and pctChgs[i-7] >= 0 and pctChgs[i] < 0 and pctChgs[i-1] < 0 and pctChgs[i-2] < 0 \312 and pctChgs[i-3] < 0 and pctChgs[i-4] < 0 and pctChgs[i-5] < 0 and pctChgs[i-6] < 0: # 账户空仓、当日下跌买入313 act = '买入'314 stock_num = account_sums[-1] // closes[i] # 买入股票数量315 account_stock = stock_num * closes[i] # 股票市值316 account_cash = account_sums[-1] - account_stock # 现金余额317 account_sum = account_cash + account_stock # 当日账户总额318 stock_cost = closes[i] # 股票成本319 account_profit = (account_sum - 1000000)*100/1000000 # 对应当日账户总收益320 elif stock_nums[-1] != 0:321 act = '卖出'322 """以收盘价卖出"""323 stock_num = 0 # 清仓,股票数量为0324 account_stock = 0 # 股票市值为0325 account_cash = account_cashes[-1] + (stock_nums[-1] * closes[i]) # 账户现金326 account_sum = account_cash + account_stock # 当日账户总额327 stock_cost = 0328 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益329 else:330 act = '空仓'331 stock_num = 0 # 空仓,股票数量为0332 account_stock = 0 # 空仓,股票市值为0333 account_cash = account_cashes[-1] # 账户现金不变334 account_sum = account_sums[-1] # 账户总额不变335 stock_cost = 0336 account_profit = (account_sum - 1000000) * 100 / 1000000 # 对应当日账户总收益337 """更新账户"""338 account_sums.append(account_sum)339 account_cashes.append(account_cash)340 account_stocks.append(account_stock)341 stock_nums.append(stock_num)342 stock_costs.append(stock_cost)343 account_profits_7.append(account_profit)344 acts.append(act)345 """绘制收益曲线"""346 # fig = plt.figure(dpi=128, figsize=(10, 6))347 # plt.plot(dates[test_start:test_end], account_profits_7, c='red')348 # plt.title(stock_code + '(n=7)')349 # fig.autofmt_xdate()350 # plt.show()351 fig = plt.figure(dpi=128, figsize=(10, 6))352 plt.plot(dates[test_start:test_end], account_profits_1, c='r')353 plt.plot(dates[test_start:test_end], account_profits_2, c='b')354 plt.plot(dates[test_start:test_end], account_profits_3, c='g')355 plt.plot(dates[test_start:test_end], account_profits_4, c='c')356 plt.plot(dates[test_start:test_end], account_profits_5, c='y')357 plt.plot(dates[test_start:test_end], account_profits_6, c='k')358 plt.plot(dates[test_start:test_end], account_profits_7, c='m')359 fig.autofmt_xdate()360 t1 = str((dates[test_start].strftime("%Y-%m-%d")))361 t2 = str(dates[test_end].strftime("%Y-%m-%d"))362 plt.title(stock_code + '(n=1~7, m=' + str(m) + ')' + t1 + '--' + t2)363 plt.legend(['n=1', 'n=2', 'n=3', 'n=4', 'n=5', 'n=6', 'n=7'], loc='best', fontsize=8)364 plt.savefig('/home/cdd/Desktop/Scraping/stocks/stock_analysis/stock_data/'365 + stock_code + stock_name + t1 + '--' + t2 + '(' + str(m) + ')' + '.png')...

Full Screen

Full Screen

test_formatters.py

Source:test_formatters.py Github

copy

Full Screen

...244 logger = StructuredLogger('test_logger')245 logger.add_handler(StreamHandler(buf, fmt))246 logger.suite_start(['test_foo', 'test_bar', 'test_baz'])247 logger.test_start('test_foo')248 logger.test_end('test_foo', 'OK')249 logger.test_start('test_bar')250 logger.test_status('test_bar', 'a subtest', 'PASS')251 logger.test_end('test_bar', 'OK')252 logger.test_start('test_baz')253 logger.test_end('test_baz', 'FAIL', 'FAIL', 'expected 0 got 1')254 logger.suite_end()255 result = buf.getvalue()256 print("Dumping result for copy/paste:")257 print(result)258 assert result == expected259@pytest.mark.parametrize("name,opts,expected", FORMATS['FAIL'],260 ids=ids('FAIL'))261def test_fail(name, opts, expected):262 stack = """263 SimpleTest.is@SimpleTest/SimpleTest.js:312:5264 @caps/tests/mochitest/test_bug246699.html:53:1265""".strip('\n')266 buf = BytesIO()267 fmt = formatters[name](**opts)268 logger = StructuredLogger('test_logger')269 logger.add_handler(StreamHandler(buf, fmt))270 logger.suite_start(['test_foo', 'test_bar', 'test_baz'])271 logger.test_start('test_foo')272 logger.test_end('test_foo', 'FAIL', 'PASS', 'expected 0 got 1')273 logger.test_start('test_bar')274 logger.test_status('test_bar', 'a subtest', 'FAIL', 'PASS', 'expected 0 got 1', stack)275 logger.test_status('test_bar', 'another subtest', 'TIMEOUT')276 logger.test_end('test_bar', 'OK')277 logger.test_start('test_baz')278 logger.test_end('test_baz', 'PASS', 'FAIL')279 logger.suite_end()280 result = buf.getvalue()281 print("Dumping result for copy/paste:")282 print(result)283 assert result == expected284@pytest.mark.parametrize("name,opts,expected", FORMATS['PRECONDITION_FAILED'],285 ids=ids('PRECONDITION_FAILED'))286def test_precondition_failed(name, opts, expected):287 buf = BytesIO()288 fmt = formatters[name](**opts)289 logger = StructuredLogger('test_logger')290 logger.add_handler(StreamHandler(buf, fmt))291 logger.suite_start(['test_foo', 'test_bar'])292 logger.test_start('test_foo')293 logger.test_end('test_foo', 'PRECONDITION_FAILED')294 logger.test_start('test_bar')295 logger.test_status('test_bar', 'a subtest', 'PASS')296 logger.test_status('test_bar', 'another subtest', 'PRECONDITION_FAILED')297 logger.test_end('test_bar', 'OK')298 logger.suite_end()299 result = buf.getvalue()300 print("Dumping result for copy/paste:")301 print(result)302 assert result == expected303@pytest.mark.parametrize("name,opts,expected", FORMATS['KNOWN-INTERMITTENT'],304 ids=ids('KNOWN-INTERMITTENT'))305def test_known_intermittent(name, opts, expected):306 buf = BytesIO()307 fmt = formatters[name](**opts)308 logger = StructuredLogger('test_logger')309 logger.add_handler(StreamHandler(buf, fmt))310 logger.suite_start(['test_foo', 'test_bar', 'test_baz'])311 logger.test_start('test_foo')312 logger.test_end('test_foo', 'FAIL', 'PASS', known_intermittent=['FAIL'])313 logger.test_start('test_bar')314 logger.test_status('test_bar', 'a subtest', 'PASS', 'FAIL',315 known_intermittent=['PASS'])316 logger.test_end('test_bar', 'OK')317 logger.test_start('test_baz')318 logger.test_end('test_baz', 'FAIL', 'FAIL', 'expected 0 got 1',319 known_intermittent=['PASS'])320 logger.suite_end()321 result = buf.getvalue()322 print("Dumping result for copy/paste:")323 print(result)324 assert result == expected325class FormatterTest(unittest.TestCase):326 def setUp(self):327 self.position = 0328 self.logger = StructuredLogger(329 "test_%s" % type(self).__name__)330 self.output_file = BytesIO()331 self.handler = StreamHandler(332 self.output_file, self.get_formatter())333 self.logger.add_handler(self.handler)334 def set_position(self, pos=None):335 if pos is None:336 pos = self.output_file.tell()337 self.position = pos338 def get_formatter(self):339 raise NotImplementedError(340 "FormatterTest subclasses must implement get_formatter")341 @property342 def loglines(self):343 self.output_file.seek(self.position)344 return [line.rstrip() for line in self.output_file.readlines()]345class TestHTMLFormatter(FormatterTest):346 def get_formatter(self):347 return HTMLFormatter()348 def test_base64_string(self):349 self.logger.suite_start([])350 self.logger.test_start("string_test")351 self.logger.test_end("string_test", "FAIL",352 extra={"data": "foobar"})353 self.logger.suite_end()354 self.assertIn(b"data:text/html;charset=utf-8;base64,Zm9vYmFy",355 self.loglines[-3])356 def test_base64_unicode(self):357 self.logger.suite_start([])358 self.logger.test_start("unicode_test")359 self.logger.test_end("unicode_test", "FAIL",360 extra={"data": unichr(0x02A9)})361 self.logger.suite_end()362 self.assertIn(b"data:text/html;charset=utf-8;base64,yqk=",363 self.loglines[-3])364 def test_base64_other(self):365 self.logger.suite_start([])366 self.logger.test_start("int_test")367 self.logger.test_end("int_test", "FAIL",368 extra={"data": {"foo": "bar"}})369 self.logger.suite_end()370 self.assertIn(b"data:text/html;charset=utf-8;base64,eyJmb28iOiAiYmFyIn0=",371 self.loglines[-3])372class TestTBPLFormatter(FormatterTest):373 def get_formatter(self):374 return TbplFormatter()375 def test_unexpected_message(self):376 self.logger.suite_start([])377 self.logger.test_start("timeout_test")378 self.logger.test_end("timeout_test",379 "TIMEOUT",380 message="timed out")381 self.assertIn(b"TEST-UNEXPECTED-TIMEOUT | timeout_test | timed out",382 self.loglines)383 self.logger.suite_end()384 def test_default_unexpected_end_message(self):385 self.logger.suite_start([])386 self.logger.test_start("timeout_test")387 self.logger.test_end("timeout_test",388 "TIMEOUT")389 self.assertIn(b"TEST-UNEXPECTED-TIMEOUT | timeout_test | expected OK",390 self.loglines)391 self.logger.suite_end()392 def test_default_unexpected_status_message(self):393 self.logger.suite_start([])394 self.logger.test_start("timeout_test")395 self.logger.test_status("timeout_test",396 "subtest",397 status="TIMEOUT")398 self.assertIn(b"TEST-UNEXPECTED-TIMEOUT | timeout_test | subtest - expected PASS",399 self.loglines)400 self.logger.test_end("timeout_test", "OK")401 self.logger.suite_end()402 def test_known_intermittent_end(self):403 self.logger.suite_start([])404 self.logger.test_start("intermittent_test")405 self.logger.test_end("intermittent_test",406 status="FAIL",407 expected="PASS",408 known_intermittent=["FAIL"])409 # test_end log format:410 # "TEST-KNOWN-INTERMITTENT-<STATUS> | <test> | took <duration>ms"411 # where duration may be different each time412 self.assertIn(b"TEST-KNOWN-INTERMITTENT-FAIL | intermittent_test | took ",413 self.loglines[2])414 self.assertIn(b"ms", self.loglines[2])415 self.logger.suite_end()416 def test_known_intermittent_status(self):417 self.logger.suite_start([])418 self.logger.test_start("intermittent_test")419 self.logger.test_status("intermittent_test",420 "subtest",421 status="FAIL",422 expected="PASS",423 known_intermittent=["FAIL"])424 self.assertIn(b"TEST-KNOWN-INTERMITTENT-FAIL | intermittent_test | subtest",425 self.loglines)426 self.logger.test_end("intermittent_test", "OK")427 self.logger.suite_end()428 def test_single_newline(self):429 self.logger.suite_start([])430 self.logger.test_start("test1")431 self.set_position()432 self.logger.test_status("test1", "subtest",433 status="PASS",434 expected="FAIL")435 self.logger.test_end("test1", "OK")436 self.logger.suite_end()437 # This sequence should not produce blanklines438 for line in self.loglines:439 self.assertNotEqual(b"", line)440 def test_process_exit(self):441 self.logger.process_exit(1234, 0)442 self.assertIn(b'TEST-INFO | 1234: exit 0', self.loglines)443 @unittest.skipUnless(os.name == 'posix', 'posix only')444 def test_process_exit_with_sig(self):445 # subprocess return code is negative when process446 # has been killed by signal on posix.447 self.logger.process_exit(1234, -signal.SIGTERM)448 self.assertIn(b'TEST-INFO | 1234: killed by SIGTERM', self.loglines)449class TestTBPLFormatterWithShutdown(FormatterTest):450 def get_formatter(self):451 return TbplFormatter(summary_on_shutdown=True)452 def test_suite_summary_on_shutdown(self):453 self.logger.suite_start([])454 self.logger.test_start("summary_test")455 self.logger.test_status("summary_test",456 "subtest",457 "FAIL",458 "PASS",459 known_intermittent=["FAIL"])460 self.logger.test_end("summary_test", "FAIL", "OK", known_intermittent=["FAIL"])461 self.logger.suite_end()462 self.logger.shutdown()463 self.assertIn(b"suite 1: 2/2 (2 known intermittent tests)", self.loglines)464 self.assertIn(b"Known Intermittent tests:", self.loglines)465 self.assertIn(b"TEST-KNOWN-INTERMITTENT-FAIL | summary_test | subtest", self.loglines)466class TestMachFormatter(FormatterTest):467 def get_formatter(self):468 return MachFormatter(disable_colors=True)469 def test_summary(self):470 self.logger.suite_start([])471 # Some tests that pass472 self.logger.test_start("test1")473 self.logger.test_end("test1", status="PASS", expected="PASS")474 self.logger.test_start("test2")475 self.logger.test_end("test2", status="PASS", expected="TIMEOUT")476 self.logger.test_start("test3")477 self.logger.test_end("test3", status="FAIL", expected="PASS")478 self.set_position()479 self.logger.suite_end()480 self.assertIn(b"Ran 3 checks (3 tests)", self.loglines)481 self.assertIn(b"Expected results: 1", self.loglines)482 self.assertIn(b"""483Unexpected results: 2484 test: 2 (1 fail, 1 pass)485""".strip(), b"\n".join(self.loglines))486 self.assertNotIn(b"test1", self.loglines)487 self.assertIn(b"UNEXPECTED-PASS test2", self.loglines)488 self.assertIn(b"FAIL test3", self.loglines)489 def test_summary_subtests(self):490 self.logger.suite_start([])491 self.logger.test_start("test1")492 self.logger.test_status("test1", "subtest1", status="PASS")493 self.logger.test_status("test1", "subtest2", status="FAIL")494 self.logger.test_end("test1", status="OK", expected="OK")495 self.logger.test_start("test2")496 self.logger.test_status("test2", "subtest1",497 status="TIMEOUT", expected="PASS")498 self.logger.test_end("test2", status="TIMEOUT", expected="OK")499 self.set_position()500 self.logger.suite_end()501 self.assertIn(b"Ran 5 checks (3 subtests, 2 tests)", self.loglines)502 self.assertIn(b"Expected results: 2", self.loglines)503 self.assertIn(b"""504Unexpected results: 3505 test: 1 (1 timeout)506 subtest: 2 (1 fail, 1 timeout)507""".strip(), b"\n".join(self.loglines))508 def test_summary_ok(self):509 self.logger.suite_start([])510 self.logger.test_start("test1")511 self.logger.test_status("test1", "subtest1", status="PASS")512 self.logger.test_status("test1", "subtest2", status="PASS")513 self.logger.test_end("test1", status="OK", expected="OK")514 self.logger.test_start("test2")515 self.logger.test_status("test2", "subtest1",516 status="PASS", expected="PASS")517 self.logger.test_end("test2", status="OK", expected="OK")518 self.set_position()519 self.logger.suite_end()520 self.assertIn(b"OK", self.loglines)521 self.assertIn(b"Expected results: 5", self.loglines)522 self.assertIn(b"Unexpected results: 0", self.loglines)523 def test_process_start(self):524 self.logger.process_start(1234)525 self.assertIn(b"Started process `1234`", self.loglines[0])526 def test_process_start_with_command(self):527 self.logger.process_start(1234, command='test cmd')528 self.assertIn(b"Started process `1234` (test cmd)", self.loglines[0])529 def test_process_exit(self):530 self.logger.process_exit(1234, 0)531 self.assertIn(b'1234: exit 0', self.loglines[0])532 @unittest.skipUnless(os.name == 'posix', 'posix only')533 def test_process_exit_with_sig(self):534 # subprocess return code is negative when process535 # has been killed by signal on posix.536 self.logger.process_exit(1234, -signal.SIGTERM)537 self.assertIn(b'1234: killed by SIGTERM', self.loglines[0])538class TestGroupingFormatter(FormatterTest):539 def get_formatter(self):540 return GroupingFormatter()541 def test_results_total(self):542 self.logger.suite_start([])543 self.logger.test_start("test1")544 self.logger.test_status("test1", "subtest1", status="PASS")545 self.logger.test_status("test1", "subtest1", status="PASS")546 self.logger.test_end("test1", status="OK")547 self.logger.test_start("test2")548 self.logger.test_status("test2",549 "subtest2",550 status="FAIL",551 expected="PASS",552 known_intermittent=["FAIL"])553 self.logger.test_end("test2", status="FAIL", expected="OK")554 self.set_position()555 self.logger.suite_end()556 self.assertIn(b"Ran 2 tests finished in 0.0 seconds.", self.loglines)557 self.assertIn(b" \xe2\x80\xa2 1 ran as expected. 0 tests skipped.",558 self.loglines)559 self.assertIn(b" \xe2\x80\xa2 1 known intermittent results.", self.loglines)560 self.assertIn(b" \xe2\x80\xa2 1 tests failed unexpectedly", self.loglines)561 self.assertIn(b" \xe2\x96\xb6 FAIL [expected OK] test2", self.loglines)562 self.assertIn(b"""563 \xe2\x96\xb6 FAIL [expected PASS, known intermittent [FAIL] test2, subtest2564""".strip(b"\n"), self.loglines)565class TestXUnitFormatter(FormatterTest):566 def get_formatter(self):567 return XUnitFormatter()568 def log_as_xml(self):569 return ET.fromstring(b'\n'.join(self.loglines))570 def test_stacktrace_is_present(self):571 self.logger.suite_start([])572 self.logger.test_start("test1")573 self.logger.test_end(574 "test1", "fail", message="Test message", stack='this\nis\na\nstack')575 self.logger.suite_end()576 root = self.log_as_xml()577 self.assertIn('this\nis\na\nstack', root.find('testcase/failure').text)578 def test_failure_message(self):579 self.logger.suite_start([])580 self.logger.test_start("test1")581 self.logger.test_end("test1", "fail", message="Test message")582 self.logger.suite_end()583 root = self.log_as_xml()584 self.assertEquals('Expected OK, got FAIL', root.find(585 'testcase/failure').get('message'))586 def test_suite_attrs(self):587 self.logger.suite_start([])588 self.logger.test_start("test1")589 self.logger.test_end("test1", "ok", message="Test message")590 self.logger.suite_end()591 root = self.log_as_xml()592 self.assertEqual(root.get('skips'), '0')593 self.assertEqual(root.get('failures'), '0')594 self.assertEqual(root.get('errors'), '0')595 self.assertEqual(root.get('tests'), '1')596 self.assertEqual(root.get('time'), '0.00')597 def test_time_is_not_rounded(self):598 # call formatter directly, it is easier here599 formatter = self.get_formatter()600 formatter.suite_start(dict(time=55000))601 formatter.test_start(dict(time=55100))602 formatter.test_end(603 dict(time=55558, test='id', message='message', status='PASS'))604 xml_string = formatter.suite_end(dict(time=55559))605 root = ET.fromstring(xml_string)606 self.assertEqual(root.get('time'), '0.56')607 self.assertEqual(root.find('testcase').get('time'), '0.46')608if __name__ == '__main__':...

Full Screen

Full Screen

stock_act.py

Source:stock_act.py Github

copy

Full Screen

1import pandas as pd2import numpy as np3from pandas import DataFrame4import matplotlib.pyplot as plt5from .stock_list import *6from .stock_analysis import *7from .stock_analysis2 import *8from .stock_analysis3 import *9from .stock_analysis4_1 import *10from .stock_analysis4_2 import *11from .stock_analysis4_3 import *12from .stock_analysis4_4 import *13from .stock_analysis4_5 import *14from .stock_analysis4_6 import *15from .stock_analysis4_7 import *16from matplotlib import pyplot as plt17stock_code = 'sh.000001'18stock_name = '上证综指'19start_time = '2010-01-01'20test_start = 24721#22# m = 1.0423#24# get_stockdata(stock_code, stock_name, start_time)25# get_data(stock_code, stock_name)26# # stock_analysis_1(stock_code, stock_name, test_start, test_end, m)27# # stock_analysis_2(stock_code, stock_name, test_start, test_end, m)28# # stock_analysis_3(stock_code, stock_name, test_start, test_end, m)29# # stock_analysis_4(stock_code, stock_name, test_start, test_end, m)30# # stock_analysis_5(stock_code, stock_name, test_start, test_end, m)31# # stock_analysis_6(stock_code, stock_name, test_start, test_end, m)32# # stock_analysis_7(stock_code, stock_name, test_start, test_end, m)33#34# for i in range(9, 10):35# test_end = test_start + 247 * i36# get_data_analysis4_7(stock_code, stock_name, test_start, test_end)37 # stock_analysis_1(stock_code, stock_name, test_start, test_end, m)38list_n1 = [-29.44, -50.66, -34.47, -27.58, -29.63, -29.44, -29.44, -29.44]39list_n2 = [9.53, -20.71, 1.4, 6.35, 11.05, 14.99, 9.78, 9.53]40list_n3 = [-7.62, -21.97, -15.66, -11.52, -7.83, -5.76, -4.42, -3.51]41list_n4 = [16.28, 9.59, 17.32, 20.01, 22.27, 16.28, 16.28, 16.28]42list_n5 = [9.41, 0.36, 5.35, 6.97, 8.01, 9.05, 9.41, 9.41]43list_n6 = [2.13, 1.63, 2.11, 2.13, 2.13, 2.13, 2.13, 2.13]44list_n7 = [-1.13, -1.95, -1.13, -1.13, -1.13, -1.13, -1.13, -1.13]45list_n = [0, 1, 2, 3, 4, 5, 6, 7]46list_act1 = [0.54, 0.55, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54]47list_act2 = [0.52, 0.54, 0.53, 0.52, 0.52, 0.52, 0.52, 0.52]48list_act3 = [0.52, 0.53, 0.52, 0.52, 0.52, 0.52, 0.52, 0.52]49list_act4 = [0.6, 0.62, 0.62, 0.62, 0.62, 0.6, 0.6, 0.6]50list_act5 = [0.58, 0.58, 0.58, 0.58, 0.58, 0.58, 0.58, 0.58]51list_act6 = [0.5, 0.6, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5]52list_act7 = [0.4, 0.4, 0.4, 0.4, 0.4, 0.4, 0.4, 0.4]53list_act = [0, 1, 2, 3, 4, 5, 6, 7]54# fig = plt.figure(dpi=128, figsize=(10, 6))55# plt.plot(list_n, list_n1, c='b')56# plt.plot(list_n, list_n2, c='g')57# plt.plot(list_n, list_n3, c='r')58# plt.plot(list_n, list_n4, c='c')59# plt.plot(list_n, list_n5, c='m')60# plt.plot(list_n, list_n6, c='y')61# plt.plot(list_n, list_n7, c='k')62# plt.legend(['n=1', 'n=2', 'n=3', 'n=4', 'n=5', 'n=6', 'n=7'], loc='best', fontsize=5)63#64# plt.show()65#66# np_nm = np.array([list_n1, list_n2, list_n3, list_n4, list_n5, list_n6, list_n7])67# print(np_nm)68# pd_nm = DataFrame(data=np_nm,69# index=['n=1', 'n=2', 'n=3', 'n=4', 'n=5', 'n=6', 'n=7'],70# columns=['m=1', 'm=1.01', 'm=1.02', 'm=1.03', 'm=1.04', 'm=1.05', 'm=1.06', 'm=1.07'])71#72# print(pd_nm)73fig = plt.figure(dpi=128, figsize=(10, 6))74plt.plot(list_act, list_act1, c='b')75plt.plot(list_act, list_act2, c='g')76plt.plot(list_act, list_act3, c='r')77plt.plot(list_act, list_act4, c='c')78plt.plot(list_act, list_act5, c='m')79plt.plot(list_act, list_act6, c='y')80plt.plot(list_act, list_act7, c='k')81plt.legend(['n=1', 'n=2', 'n=3', 'n=4', 'n=5', 'n=6', 'n=7'], loc='best', fontsize=5)82plt.savefig('/home/cdd/Desktop/Scraping/stocks/stock_analysis/stock_data2/'83 + stock_code + stock_name + '(n=7)' + '.png')...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful