| 156 | split_options("auth=GSSAPI") |
| 157 | |
| 158 | def test_parse_uri(self): |
| 159 | self.assertRaises(InvalidURI, parse_uri, "http://foobar.com") |
| 160 | self.assertRaises(InvalidURI, parse_uri, "http://foo@foobar.com") |
| 161 | self.assertRaises(ValueError, parse_uri, "mongodb://::1", 27017) |
| 162 | |
| 163 | orig: dict = { |
| 164 | "nodelist": [("localhost", 27017)], |
| 165 | "username": None, |
| 166 | "password": None, |
| 167 | "database": None, |
| 168 | "collection": None, |
| 169 | "options": {}, |
| 170 | "fqdn": None, |
| 171 | } |
| 172 | |
| 173 | res: dict = copy.deepcopy(orig) |
| 174 | self.assertEqual(res, parse_uri("mongodb://localhost")) |
| 175 | |
| 176 | res.update({"username": "fred", "password": "foobar"}) |
| 177 | self.assertEqual(res, parse_uri("mongodb://fred:foobar@localhost")) |
| 178 | |
| 179 | res.update({"database": "baz"}) |
| 180 | self.assertEqual(res, parse_uri("mongodb://fred:foobar@localhost/baz")) |
| 181 | |
| 182 | res = copy.deepcopy(orig) |
| 183 | res["nodelist"] = [("example1.com", 27017), ("example2.com", 27017)] |
| 184 | self.assertEqual(res, parse_uri("mongodb://example1.com:27017,example2.com:27017")) |
| 185 | |
| 186 | res = copy.deepcopy(orig) |
| 187 | res["nodelist"] = [("localhost", 27017), ("localhost", 27018), ("localhost", 27019)] |
| 188 | self.assertEqual(res, parse_uri("mongodb://localhost,localhost:27018,localhost:27019")) |
| 189 | |
| 190 | res = copy.deepcopy(orig) |
| 191 | res["database"] = "foo" |
| 192 | self.assertEqual(res, parse_uri("mongodb://localhost/foo")) |
| 193 | |
| 194 | res = copy.deepcopy(orig) |
| 195 | self.assertEqual(res, parse_uri("mongodb://localhost/")) |
| 196 | |
| 197 | res.update({"database": "test", "collection": "yield_historical.in"}) |
| 198 | self.assertEqual(res, parse_uri("mongodb://localhost/test.yield_historical.in")) |
| 199 | |
| 200 | res.update({"username": "fred", "password": "foobar"}) |
| 201 | self.assertEqual(res, parse_uri("mongodb://fred:foobar@localhost/test.yield_historical.in")) |
| 202 | |
| 203 | res = copy.deepcopy(orig) |
| 204 | res["nodelist"] = [("example1.com", 27017), ("example2.com", 27017)] |
| 205 | res.update({"database": "test", "collection": "yield_historical.in"}) |
| 206 | self.assertEqual( |
| 207 | res, |
| 208 | parse_uri("mongodb://example1.com:27017,example2.com:27017/test.yield_historical.in"), |
| 209 | ) |
| 210 | |
| 211 | # Test socket path without escaped characters. |
| 212 | self.assertRaises(InvalidURI, parse_uri, "mongodb:///tmp/mongodb-27017.sock") |
| 213 | |
| 214 | # Test with escaped characters. |
| 215 | res = copy.deepcopy(orig) |