(self)
| 95 | |
| 96 | def create_test(test, test_workdir): |
| 97 | def run_scenario(self): |
| 98 | compressors = (test.get("options") or {}).get("compressors", []) |
| 99 | if "snappy" in compressors and not _have_snappy(): |
| 100 | self.skipTest("This test needs the snappy module.") |
| 101 | valid = True |
| 102 | warning = False |
| 103 | expected_warning = test.get("warning", False) |
| 104 | expected_valid = test.get("valid", True) |
| 105 | |
| 106 | with warnings.catch_warnings(record=True) as ctx: |
| 107 | warnings.simplefilter("ignore", category=ResourceWarning) |
| 108 | try: |
| 109 | options = parse_uri(test["uri"], warn=True) |
| 110 | except Exception: |
| 111 | valid = False |
| 112 | else: |
| 113 | warning = len(ctx) > 0 |
| 114 | if expected_valid and warning and not expected_warning: |
| 115 | raise ValueError("Got unexpected warning(s): ", [str(i) for i in ctx]) |
| 116 | |
| 117 | self.assertEqual( |
| 118 | valid, |
| 119 | expected_valid, |
| 120 | get_error_message_template(not expected_valid, "error") % test["description"], |
| 121 | ) |
| 122 | |
| 123 | if expected_valid: |
| 124 | self.assertEqual( |
| 125 | warning, |
| 126 | expected_warning, |
| 127 | get_error_message_template(expected_warning, "warning") % test["description"], |
| 128 | ) |
| 129 | |
| 130 | # Compare hosts and port. |
| 131 | if test["hosts"] is not None: |
| 132 | self.assertEqual( |
| 133 | len(test["hosts"]), |
| 134 | len(options["nodelist"]), |
| 135 | "Incorrect number of hosts parsed from URI", |
| 136 | ) |
| 137 | |
| 138 | for exp, actual in zip(test["hosts"], options["nodelist"]): |
| 139 | self.assertEqual( |
| 140 | exp["host"], |
| 141 | actual[0], |
| 142 | "Expected host {} but got {}".format(exp["host"], actual[0]), |
| 143 | ) |
| 144 | if exp["port"] is not None: |
| 145 | self.assertEqual( |
| 146 | exp["port"], |
| 147 | actual[1], |
| 148 | "Expected port {} but got {}".format(exp["port"], actual), |
| 149 | ) |
| 150 | |
| 151 | # Compare auth options. |
| 152 | auth = test["auth"] |
| 153 | if auth is not None: |
| 154 | auth["database"] = auth.pop("db") # db == database |
nothing calls this directly
no test coverage detected