mirror of
https://github.com/kennethreitz/pydantic.git
synced 2026-06-05 23:00:18 +00:00
8d98c499df
* url parsing in pydantic-core * fix tests * fix mypy tests
698 lines
25 KiB
Python
698 lines
25 KiB
Python
import pytest
|
|
from pydantic_core import PydanticCustomError, Url
|
|
from typing_extensions import Annotated
|
|
|
|
from pydantic import (
|
|
AmqpDsn,
|
|
AnyUrl,
|
|
BaseModel,
|
|
CockroachDsn,
|
|
FileUrl,
|
|
HttpUrl,
|
|
KafkaDsn,
|
|
MongoDsn,
|
|
NameEmail,
|
|
PostgresDsn,
|
|
RedisDsn,
|
|
Strict,
|
|
UrlConstraints,
|
|
ValidationError,
|
|
)
|
|
from pydantic.networks import validate_email
|
|
|
|
try:
|
|
import email_validator
|
|
except ImportError:
|
|
email_validator = None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'value',
|
|
[
|
|
'http://example.org',
|
|
'http://test',
|
|
'http://localhost',
|
|
'https://example.org/whatever/next/',
|
|
'postgres://user:pass@localhost:5432/app',
|
|
'postgres://just-user@localhost:5432/app',
|
|
'postgresql+asyncpg://user:pass@localhost:5432/app',
|
|
'postgresql+pg8000://user:pass@localhost:5432/app',
|
|
'postgresql+psycopg://postgres:postgres@localhost:5432/hatch',
|
|
'postgresql+psycopg2://postgres:postgres@localhost:5432/hatch',
|
|
'postgresql+psycopg2cffi://user:pass@localhost:5432/app',
|
|
'postgresql+py-postgresql://user:pass@localhost:5432/app',
|
|
'postgresql+pygresql://user:pass@localhost:5432/app',
|
|
'foo-bar://example.org',
|
|
'foo.bar://example.org',
|
|
'foo0bar://example.org',
|
|
'https://example.org',
|
|
'http://localhost',
|
|
'http://localhost/',
|
|
'http://localhost:8000',
|
|
'http://localhost:8000/',
|
|
'https://foo_bar.example.com/',
|
|
'ftp://example.org',
|
|
'ftps://example.org',
|
|
'http://example.co.jp',
|
|
'http://www.example.com/a%C2%B1b',
|
|
'http://www.example.com/~username/',
|
|
'http://info.example.com?fred',
|
|
'http://info.example.com/?fred',
|
|
'http://xn--mgbh0fb.xn--kgbechtv/',
|
|
'http://example.com/blue/red%3Fand+green',
|
|
'http://www.example.com/?array%5Bkey%5D=value',
|
|
'http://xn--rsum-bpad.example.org/',
|
|
'http://123.45.67.8/',
|
|
'http://123.45.67.8:8329/',
|
|
'http://[2001:db8::ff00:42]:8329',
|
|
'http://[2001::1]:8329',
|
|
'http://[2001:db8::1]/',
|
|
'http://www.example.com:8000/foo',
|
|
'http://www.cwi.nl:80/%7Eguido/Python.html',
|
|
'https://www.python.org/путь',
|
|
'http://андрей@example.com',
|
|
# AnyUrl('https://example.com', scheme='https', host='example.com'),
|
|
'https://exam_ple.com/',
|
|
'http://twitter.com/@handle/',
|
|
'http://11.11.11.11.example.com/action',
|
|
'http://abc.11.11.11.11.example.com/action',
|
|
'http://example#',
|
|
'http://example/#',
|
|
'http://example/#fragment',
|
|
'http://example/?#',
|
|
'http://example.org/path#',
|
|
'http://example.org/path#fragment',
|
|
'http://example.org/path?query#',
|
|
'http://example.org/path?query#fragment',
|
|
'file://localhost/foo/bar',
|
|
],
|
|
)
|
|
def test_any_url_success(value):
|
|
class Model(BaseModel):
|
|
v: AnyUrl
|
|
|
|
assert Model(v=value).v, value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'value,err_type,err_msg',
|
|
[
|
|
('http:///', 'url_parsing', 'Input should be a valid URL, empty host'),
|
|
('http://??', 'url_parsing', 'Input should be a valid URL, empty host'),
|
|
(
|
|
'https://example.org more',
|
|
'url_parsing',
|
|
'Input should be a valid URL, invalid domain character',
|
|
),
|
|
('$https://example.org', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('../icons/logo.gif', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('abc', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('..', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('/', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('+http://example.com/', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('ht*tp://example.com/', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
(' ', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
('', 'url_parsing', 'Input should be a valid URL, relative URL without a base'),
|
|
(None, 'url_type', 'URL input should be a string or URL'),
|
|
(
|
|
'http://2001:db8::ff00:42:8329',
|
|
'url_parsing',
|
|
'Input should be a valid URL, invalid port number',
|
|
),
|
|
('http://[192.168.1.1]:8329', 'url_parsing', 'Input should be a valid URL, invalid IPv6 address'),
|
|
('http://example.com:99999', 'url_parsing', 'Input should be a valid URL, invalid port number'),
|
|
],
|
|
)
|
|
def test_any_url_invalid(value, err_type, err_msg):
|
|
class Model(BaseModel):
|
|
v: AnyUrl
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(v=value)
|
|
assert len(exc_info.value.errors()) == 1, exc_info.value.errors()
|
|
error = exc_info.value.errors()[0]
|
|
# debug(error)
|
|
assert {'type': error['type'], 'msg': error['msg']} == {'type': err_type, 'msg': err_msg}
|
|
|
|
|
|
def validate_url(s):
|
|
class Model(BaseModel):
|
|
v: AnyUrl
|
|
|
|
return Model(v=s).v
|
|
|
|
|
|
def test_any_url_parts():
|
|
url = validate_url('http://example.org')
|
|
assert str(url) == 'http://example.org/'
|
|
assert repr(url) == "Url('http://example.org/')"
|
|
assert url.scheme == 'http'
|
|
assert url.host == 'example.org'
|
|
assert url.port == 80
|
|
|
|
|
|
def test_url_repr():
|
|
url = validate_url('http://user:password@example.org:1234/the/path/?query=here#fragment=is;this=bit')
|
|
assert str(url) == 'http://user:password@example.org:1234/the/path/?query=here#fragment=is;this=bit'
|
|
assert repr(url) == "Url('http://user:password@example.org:1234/the/path/?query=here#fragment=is;this=bit')"
|
|
assert url.scheme == 'http'
|
|
assert url.username == 'user'
|
|
assert url.password == 'password'
|
|
assert url.host == 'example.org'
|
|
assert url.port == 1234
|
|
assert url.path == '/the/path/'
|
|
assert url.query == 'query=here'
|
|
assert url.fragment == 'fragment=is;this=bit'
|
|
|
|
|
|
def test_ipv4_port():
|
|
url = validate_url('ftp://123.45.67.8:8329/')
|
|
assert url.scheme == 'ftp'
|
|
assert url.host == '123.45.67.8'
|
|
assert url.port == 8329
|
|
assert url.username is None
|
|
assert url.password is None
|
|
|
|
|
|
def test_ipv4_no_port():
|
|
url = validate_url('ftp://123.45.67.8')
|
|
assert url.scheme == 'ftp'
|
|
assert url.host == '123.45.67.8'
|
|
assert url.port == 21
|
|
assert url.username is None
|
|
assert url.password is None
|
|
|
|
|
|
def test_ipv6_port():
|
|
url = validate_url('wss://[2001:db8::ff00:42]:8329')
|
|
assert url.scheme == 'wss'
|
|
assert url.host == '[2001:db8::ff00:42]'
|
|
assert url.port == 8329
|
|
|
|
|
|
def test_int_domain():
|
|
url = validate_url('https://£££.org')
|
|
assert url.host == 'xn--9aaa.org'
|
|
assert str(url) == 'https://xn--9aaa.org/'
|
|
|
|
|
|
def test_co_uk():
|
|
url = validate_url('http://example.co.uk')
|
|
assert str(url) == 'http://example.co.uk/'
|
|
assert url.scheme == 'http'
|
|
assert url.host == 'example.co.uk'
|
|
|
|
|
|
def test_user_no_password():
|
|
url = validate_url('http://user:@example.org')
|
|
assert url.username == 'user'
|
|
assert url.password is None
|
|
assert url.host == 'example.org'
|
|
|
|
|
|
def test_user_info_no_user():
|
|
url = validate_url('http://:password@example.org')
|
|
assert url.username is None
|
|
assert url.password == 'password'
|
|
assert url.host == 'example.org'
|
|
|
|
|
|
def test_at_in_path():
|
|
url = validate_url('https://twitter.com/@handle')
|
|
assert url.scheme == 'https'
|
|
assert url.host == 'twitter.com'
|
|
assert url.username is None
|
|
assert url.password is None
|
|
assert url.path == '/@handle'
|
|
|
|
|
|
def test_fragment_without_query():
|
|
url = validate_url('https://pydantic-docs.helpmanual.io/usage/types/#constrained-types')
|
|
assert url.scheme == 'https'
|
|
assert url.host == 'pydantic-docs.helpmanual.io'
|
|
assert url.path == '/usage/types/'
|
|
assert url.query is None
|
|
assert url.fragment == 'constrained-types'
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'value,expected',
|
|
[
|
|
('http://example.org', 'http://example.org/'),
|
|
('http://example.org/foobar', 'http://example.org/foobar'),
|
|
('http://example.org.', 'http://example.org./'),
|
|
('http://example.org./foobar', 'http://example.org./foobar'),
|
|
('HTTP://EXAMPLE.ORG', 'http://example.org/'),
|
|
('https://example.org', 'https://example.org/'),
|
|
('https://example.org?a=1&b=2', 'https://example.org/?a=1&b=2'),
|
|
('https://example.org#a=3;b=3', 'https://example.org/#a=3;b=3'),
|
|
('https://foo_bar.example.com/', 'https://foo_bar.example.com/'),
|
|
('https://exam_ple.com/', 'https://exam_ple.com/'),
|
|
('https://example.xn--p1ai', 'https://example.xn--p1ai/'),
|
|
('https://example.xn--vermgensberatung-pwb', 'https://example.xn--vermgensberatung-pwb/'),
|
|
('https://example.xn--zfr164b', 'https://example.xn--zfr164b/'),
|
|
],
|
|
)
|
|
def test_http_url_success(value, expected):
|
|
class Model(BaseModel):
|
|
v: HttpUrl
|
|
|
|
assert str(Model(v=value).v) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'value,err_type,err_msg',
|
|
[
|
|
(
|
|
'ftp://example.com/',
|
|
'url_scheme',
|
|
"URL scheme should be 'http' or 'https'",
|
|
),
|
|
(
|
|
'x' * 2084,
|
|
'url_too_long',
|
|
'URL should have at most 2083 characters',
|
|
),
|
|
],
|
|
)
|
|
def test_http_url_invalid(value, err_type, err_msg):
|
|
class Model(BaseModel):
|
|
v: HttpUrl
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(v=value)
|
|
assert len(exc_info.value.errors()) == 1, exc_info.value.errors()
|
|
error = exc_info.value.errors()[0]
|
|
assert {'type': error['type'], 'msg': error['msg']} == {'type': err_type, 'msg': err_msg}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'input,output',
|
|
[
|
|
(' https://www.example.com \n', 'https://www.example.com/'),
|
|
(b'https://www.example.com', 'https://www.example.com/'),
|
|
# https://www.xudongz.com/blog/2017/idn-phishing/ accepted but converted
|
|
('https://www.аррӏе.com/', 'https://www.xn--80ak6aa92e.com/'),
|
|
('https://exampl£e.org', 'https://xn--example-gia.org/'),
|
|
('https://example.珠宝', 'https://example.xn--pbt977c/'),
|
|
('https://example.vermögensberatung', 'https://example.xn--vermgensberatung-pwb/'),
|
|
('https://example.рф', 'https://example.xn--p1ai/'),
|
|
('https://exampl£e.珠宝', 'https://xn--example-gia.xn--pbt977c/'),
|
|
],
|
|
)
|
|
def test_coerce_url(input, output):
|
|
class Model(BaseModel):
|
|
v: HttpUrl
|
|
|
|
assert str(Model(v=input).v) == output
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'value,expected',
|
|
[
|
|
('file:///foo/bar', 'file:///foo/bar'),
|
|
('file://localhost/foo/bar', 'file:///foo/bar'),
|
|
('file:////localhost/foo/bar', 'file:///localhost/foo/bar'),
|
|
],
|
|
)
|
|
def test_file_url_success(value, expected):
|
|
class Model(BaseModel):
|
|
v: FileUrl
|
|
|
|
assert str(Model(v=value).v) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'url,expected_port, expected_str',
|
|
[
|
|
('https://www.example.com/', 443, 'https://www.example.com/'),
|
|
('https://www.example.com:443/', 443, 'https://www.example.com/'),
|
|
('https://www.example.com:8089/', 8089, 'https://www.example.com:8089/'),
|
|
('http://www.example.com/', 80, 'http://www.example.com/'),
|
|
('http://www.example.com:80/', 80, 'http://www.example.com/'),
|
|
('http://www.example.com:8080/', 8080, 'http://www.example.com:8080/'),
|
|
],
|
|
)
|
|
def test_http_urls_default_port(url, expected_port, expected_str):
|
|
class Model(BaseModel):
|
|
v: HttpUrl
|
|
|
|
m = Model(v=url)
|
|
assert m.v.port == expected_port
|
|
assert str(m.v) == expected_str
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'dsn',
|
|
[
|
|
'postgres://user:pass@localhost:5432/app',
|
|
'postgresql://user:pass@localhost:5432/app',
|
|
'postgresql+asyncpg://user:pass@localhost:5432/app',
|
|
'postgres://user:pass@host1.db.net,host2.db.net:6432/app',
|
|
],
|
|
)
|
|
def test_postgres_dsns(dsn):
|
|
class Model(BaseModel):
|
|
a: PostgresDsn
|
|
|
|
assert str(Model(a=dsn).a) == dsn
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'dsn,error_message',
|
|
(
|
|
(
|
|
'postgres://user:pass@host1.db.net:4321,/foo/bar:5432/app',
|
|
{
|
|
'type': 'url_parsing',
|
|
'loc': ('a',),
|
|
'msg': 'Input should be a valid URL, empty host',
|
|
'input': 'postgres://user:pass@host1.db.net:4321,/foo/bar:5432/app',
|
|
},
|
|
),
|
|
(
|
|
'postgres://user:pass@host1.db.net,/app',
|
|
{
|
|
'type': 'url_parsing',
|
|
'loc': ('a',),
|
|
'msg': 'Input should be a valid URL, empty host',
|
|
'input': 'postgres://user:pass@host1.db.net,/app',
|
|
},
|
|
),
|
|
(
|
|
'postgres://user:pass@/foo/bar:5432,host1.db.net:4321/app',
|
|
{
|
|
'type': 'url_parsing',
|
|
'loc': ('a',),
|
|
'msg': 'Input should be a valid URL, empty host',
|
|
'input': 'postgres://user:pass@/foo/bar:5432,host1.db.net:4321/app',
|
|
},
|
|
),
|
|
(
|
|
'postgres://user@/foo/bar:5432/app',
|
|
{
|
|
'type': 'url_parsing',
|
|
'loc': ('a',),
|
|
'msg': 'Input should be a valid URL, empty host',
|
|
'input': 'postgres://user@/foo/bar:5432/app',
|
|
},
|
|
),
|
|
(
|
|
'http://example.org',
|
|
{
|
|
'type': 'url_scheme',
|
|
'loc': ('a',),
|
|
'msg': (
|
|
"URL scheme should be 'postgres', 'postgresql', 'postgresql+asyncpg', 'postgresql+pg8000', "
|
|
"'postgresql+psycopg', 'postgresql+psycopg2', 'postgresql+psycopg2cffi', "
|
|
"'postgresql+py-postgresql' or 'postgresql+pygresql'"
|
|
),
|
|
'input': 'http://example.org',
|
|
},
|
|
),
|
|
),
|
|
)
|
|
def test_postgres_dsns_validation_error(dsn, error_message):
|
|
class Model(BaseModel):
|
|
a: PostgresDsn
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a=dsn)
|
|
error = exc_info.value.errors()[0]
|
|
error.pop('ctx', None)
|
|
assert error == error_message
|
|
|
|
|
|
def test_multihost_postgres_dsns():
|
|
class Model(BaseModel):
|
|
a: PostgresDsn
|
|
|
|
any_multihost_url = Model(a='postgres://user:pass@host1.db.net:4321,host2.db.net:6432/app').a
|
|
assert str(any_multihost_url) == 'postgres://user:pass@host1.db.net:4321,host2.db.net:6432/app'
|
|
assert any_multihost_url.scheme == 'postgres'
|
|
assert any_multihost_url.path == '/app'
|
|
# insert_assert(any_multihost_url.hosts())
|
|
assert any_multihost_url.hosts() == [
|
|
{'username': 'user', 'password': 'pass', 'host': 'host1.db.net', 'port': 4321},
|
|
{'username': None, 'password': None, 'host': 'host2.db.net', 'port': 6432},
|
|
]
|
|
|
|
any_multihost_url = Model(a='postgres://user:pass@host.db.net:4321/app').a
|
|
assert any_multihost_url.scheme == 'postgres'
|
|
assert str(any_multihost_url) == 'postgres://user:pass@host.db.net:4321/app'
|
|
assert any_multihost_url.path == '/app'
|
|
# insert_assert(any_multihost_url.hosts())
|
|
assert any_multihost_url.hosts() == [{'username': 'user', 'password': 'pass', 'host': 'host.db.net', 'port': 4321}]
|
|
|
|
|
|
def test_cockroach_dsns():
|
|
class Model(BaseModel):
|
|
a: CockroachDsn
|
|
|
|
assert str(Model(a='cockroachdb://user:pass@localhost:5432/app').a) == 'cockroachdb://user:pass@localhost:5432/app'
|
|
assert (
|
|
str(Model(a='cockroachdb+psycopg2://user:pass@localhost:5432/app').a)
|
|
== 'cockroachdb+psycopg2://user:pass@localhost:5432/app'
|
|
)
|
|
assert (
|
|
str(Model(a='cockroachdb+asyncpg://user:pass@localhost:5432/app').a)
|
|
== 'cockroachdb+asyncpg://user:pass@localhost:5432/app'
|
|
)
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a='http://example.org')
|
|
assert exc_info.value.errors()[0]['type'] == 'url_scheme'
|
|
|
|
|
|
def test_amqp_dsns():
|
|
class Model(BaseModel):
|
|
a: AmqpDsn
|
|
|
|
m = Model(a='amqp://user:pass@localhost:1234/app')
|
|
assert str(m.a) == 'amqp://user:pass@localhost:1234/app'
|
|
assert m.a.username == 'user'
|
|
assert m.a.password == 'pass'
|
|
|
|
m = Model(a='amqps://user:pass@localhost:5432//')
|
|
assert str(m.a) == 'amqps://user:pass@localhost:5432//'
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a='http://example.org')
|
|
assert exc_info.value.errors()[0]['type'] == 'url_scheme'
|
|
|
|
# Password is not required for AMQP protocol
|
|
m = Model(a='amqp://localhost:1234/app')
|
|
assert str(m.a) == 'amqp://localhost:1234/app'
|
|
assert m.a.username is None
|
|
assert m.a.password is None
|
|
|
|
# Only schema is required for AMQP protocol.
|
|
# https://www.rabbitmq.com/uri-spec.html
|
|
m = Model(a='amqps://')
|
|
assert m.a.scheme == 'amqps'
|
|
assert m.a.host is None
|
|
assert m.a.port is None
|
|
assert m.a.path is None
|
|
|
|
|
|
def test_redis_dsns():
|
|
class Model(BaseModel):
|
|
a: RedisDsn
|
|
|
|
m = Model(a='redis://user:pass@localhost:1234/app')
|
|
assert str(m.a) == 'redis://user:pass@localhost:1234/app'
|
|
assert m.a.username == 'user'
|
|
assert m.a.password == 'pass'
|
|
|
|
m = Model(a='rediss://user:pass@localhost:1234/app')
|
|
assert str(m.a) == 'rediss://user:pass@localhost:1234/app'
|
|
|
|
m = Model(a='rediss://:pass@localhost:1234')
|
|
assert str(m.a) == 'rediss://:pass@localhost:1234/0'
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a='http://example.org')
|
|
assert exc_info.value.errors()[0]['type'] == 'url_scheme'
|
|
|
|
# Password is not required for Redis protocol
|
|
m = Model(a='redis://localhost:1234/app')
|
|
assert str(m.a) == 'redis://localhost:1234/app'
|
|
assert m.a.username is None
|
|
assert m.a.password is None
|
|
|
|
# Only schema is required for Redis protocol. Otherwise it will be set to default
|
|
# https://www.iana.org/assignments/uri-schemes/prov/redis
|
|
m = Model(a='rediss://')
|
|
assert m.a.scheme == 'rediss'
|
|
assert m.a.host == 'localhost'
|
|
assert m.a.port == 6379
|
|
assert m.a.path == '/0'
|
|
|
|
|
|
def test_mongodb_dsns():
|
|
class Model(BaseModel):
|
|
a: MongoDsn
|
|
|
|
# TODO: Need to unit tests about "Replica Set", "Sharded cluster" and other deployment modes of MongoDB
|
|
m = Model(a='mongodb://user:pass@localhost:1234/app')
|
|
assert str(m.a) == 'mongodb://user:pass@localhost:1234/app'
|
|
# insert_assert(m.a.hosts())
|
|
assert m.a.hosts() == [{'username': 'user', 'password': 'pass', 'host': 'localhost', 'port': 1234}]
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a='http://example.org')
|
|
assert exc_info.value.errors()[0]['type'] == 'url_scheme'
|
|
|
|
# Password is not required for MongoDB protocol
|
|
m = Model(a='mongodb://localhost:1234/app')
|
|
assert str(m.a) == 'mongodb://localhost:1234/app'
|
|
# insert_assert(m.a.hosts())
|
|
assert m.a.hosts() == [{'username': None, 'password': None, 'host': 'localhost', 'port': 1234}]
|
|
|
|
# Only schema and host is required for MongoDB protocol
|
|
m = Model(a='mongodb://localhost')
|
|
assert m.a.scheme == 'mongodb'
|
|
# insert_assert(m.a.hosts())
|
|
assert m.a.hosts() == [{'username': None, 'password': None, 'host': 'localhost', 'port': 27017}]
|
|
|
|
|
|
def test_kafka_dsns():
|
|
class Model(BaseModel):
|
|
a: KafkaDsn
|
|
|
|
m = Model(a='kafka://')
|
|
assert m.a.scheme == 'kafka'
|
|
assert m.a.host == 'localhost'
|
|
assert m.a.port == 9092
|
|
assert str(m.a) == 'kafka://localhost:9092'
|
|
|
|
m = Model(a='kafka://kafka1')
|
|
assert str(m.a) == 'kafka://kafka1:9092'
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
Model(a='http://example.org')
|
|
assert exc_info.value.errors()[0]['type'] == 'url_scheme'
|
|
|
|
m = Model(a='kafka://kafka3:9093')
|
|
assert m.a.username is None
|
|
assert m.a.password is None
|
|
|
|
|
|
def test_custom_schemes():
|
|
class Model(BaseModel):
|
|
v: Annotated[Url, UrlConstraints(allowed_schemes=['ws', 'wss']), Strict()]
|
|
|
|
class Model2(BaseModel):
|
|
v: Annotated[Url, UrlConstraints(host_required=False, allowed_schemes=['foo'])]
|
|
|
|
assert str(Model(v='ws://example.org').v) == 'ws://example.org/'
|
|
assert str(Model2(v='foo:///foo/bar').v) == 'foo:///foo/bar'
|
|
|
|
with pytest.raises(ValidationError, match=r"URL scheme should be 'ws' or 'wss' \[type=url_scheme,"):
|
|
Model(v='http://example.org')
|
|
|
|
with pytest.raises(ValidationError, match='leading or trailing control or space character are ignored in URLs'):
|
|
Model(v='ws://example.org ')
|
|
|
|
with pytest.raises(ValidationError, match=r'syntax rules, expected // \[type=url_syntax_violation,'):
|
|
Model(v='ws:///foo/bar')
|
|
|
|
|
|
@pytest.mark.xfail(reason='URL to JSON not yet supported')
|
|
def test_json():
|
|
class Model(BaseModel):
|
|
v: HttpUrl
|
|
|
|
m = Model(v='http://foo@example.net')
|
|
assert m.json() == '{"v": "http://foo@example.net"}'
|
|
|
|
|
|
@pytest.mark.skipif(not email_validator, reason='email_validator not installed')
|
|
@pytest.mark.parametrize(
|
|
'value,name,email',
|
|
[
|
|
('foobar@example.com', 'foobar', 'foobar@example.com'),
|
|
('s@muelcolvin.com', 's', 's@muelcolvin.com'),
|
|
('Samuel Colvin <s@muelcolvin.com>', 'Samuel Colvin', 's@muelcolvin.com'),
|
|
('foobar <foobar@example.com>', 'foobar', 'foobar@example.com'),
|
|
(' foo.bar@example.com', 'foo.bar', 'foo.bar@example.com'),
|
|
('foo.bar@example.com ', 'foo.bar', 'foo.bar@example.com'),
|
|
('foo BAR <foobar@example.com >', 'foo BAR', 'foobar@example.com'),
|
|
('FOO bar <foobar@example.com> ', 'FOO bar', 'foobar@example.com'),
|
|
(' Whatever <foobar@example.com>', 'Whatever', 'foobar@example.com'),
|
|
('Whatever < foobar@example.com>', 'Whatever', 'foobar@example.com'),
|
|
('<FOOBAR@example.com> ', 'FOOBAR', 'FOOBAR@example.com'),
|
|
('ñoñó@example.com', 'ñoñó', 'ñoñó@example.com'),
|
|
('我買@example.com', '我買', '我買@example.com'),
|
|
('甲斐黒川日本@example.com', '甲斐黒川日本', '甲斐黒川日本@example.com'),
|
|
(
|
|
'чебурашкаящик-с-апельсинами.рф@example.com',
|
|
'чебурашкаящик-с-апельсинами.рф',
|
|
'чебурашкаящик-с-апельсинами.рф@example.com',
|
|
),
|
|
('उदाहरण.परीक्ष@domain.with.idn.tld', 'उदाहरण.परीक्ष', 'उदाहरण.परीक्ष@domain.with.idn.tld'),
|
|
('foo.bar@example.com', 'foo.bar', 'foo.bar@example.com'),
|
|
('foo.bar@exam-ple.com ', 'foo.bar', 'foo.bar@exam-ple.com'),
|
|
('ιωάννης@εεττ.gr', 'ιωάννης', 'ιωάννης@εεττ.gr'),
|
|
('foobar@аррӏе.com', 'foobar', 'foobar@аррӏе.com'),
|
|
('foobar@xn--80ak6aa92e.com', 'foobar', 'foobar@аррӏе.com'),
|
|
('аррӏе@example.com', 'аррӏе', 'аррӏе@example.com'),
|
|
('xn--80ak6aa92e@example.com', 'xn--80ak6aa92e', 'xn--80ak6aa92e@example.com'),
|
|
('葉士豪@臺網中心.tw', '葉士豪', '葉士豪@臺網中心.tw'),
|
|
],
|
|
)
|
|
def test_address_valid(value, name, email):
|
|
assert validate_email(value) == (name, email)
|
|
|
|
|
|
@pytest.mark.skipif(not email_validator, reason='email_validator not installed')
|
|
@pytest.mark.parametrize(
|
|
'value,reason',
|
|
[
|
|
('@example.com', 'There must be something before the @-sign.'),
|
|
('f oo.bar@example.com', 'The email address contains invalid characters before the @-sign'),
|
|
('foobar', 'The email address is not valid. It must have exactly one @-sign.'),
|
|
('foobar@localhost', 'The domain name localhost is not valid. It should have a period.'),
|
|
('foobar@127.0.0.1', 'The domain name 127.0.0.1 is not valid. It is not within a valid top-level domain.'),
|
|
('foo.bar@exam\nple.com ', None),
|
|
('foobar <foobar@example.com', None),
|
|
('foobar@.example.com', None),
|
|
('foobar@.com', None),
|
|
('foo bar@example.com', None),
|
|
('foo@bar@example.com', None),
|
|
('\n@example.com', None),
|
|
('\r@example.com', None),
|
|
('\f@example.com', None),
|
|
(' @example.com', None),
|
|
('\u0020@example.com', None),
|
|
('\u001f@example.com', None),
|
|
('"@example.com', None),
|
|
('\"@example.com', None),
|
|
(',@example.com', None),
|
|
('foobar <foobar<@example.com>', None),
|
|
('foobar <foobar@example.com>>', None),
|
|
('foobar <<foobar<@example.com>', None),
|
|
('foobar <>', None),
|
|
],
|
|
)
|
|
def test_address_invalid(value, reason):
|
|
with pytest.raises(PydanticCustomError, match=f'value is not a valid email address: {reason or ""}'):
|
|
validate_email(value)
|
|
|
|
|
|
@pytest.mark.skipif(email_validator, reason='email_validator is installed')
|
|
def test_email_validator_not_installed():
|
|
with pytest.raises(ImportError):
|
|
validate_email('s@muelcolvin.com')
|
|
|
|
|
|
@pytest.mark.skipif(not email_validator, reason='email_validator not installed')
|
|
def test_name_email():
|
|
class Model(BaseModel):
|
|
v: NameEmail
|
|
|
|
assert str(Model(v=NameEmail('foo bar', 'foobaR@example.com')).v) == 'foo bar <foobaR@example.com>'
|
|
assert str(Model(v='foo bar <foobaR@example.com>').v) == 'foo bar <foobaR@example.com>'
|
|
assert NameEmail('foo bar', 'foobaR@example.com') == NameEmail('foo bar', 'foobaR@example.com')
|
|
assert NameEmail('foo bar', 'foobaR@example.com') != NameEmail('foo bar', 'different@example.com')
|