Refactor old tests
parent
39f3f82970
commit
8b130f0361
|
@ -54,11 +54,11 @@ class HostnameValueError(Exception):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class IpAddresValueError(Exception):
|
class IpAddresValueError(ValueError):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def validate_destination_address(url):
|
def validate_private_url(url):
|
||||||
host = urlparse(url).hostname
|
host = urlparse(url).hostname
|
||||||
port = urlparse(url).port
|
port = urlparse(url).port
|
||||||
|
|
||||||
|
@ -74,5 +74,3 @@ def validate_destination_address(url):
|
||||||
raise IpAddresValueError(_("IP Address error"))
|
raise IpAddresValueError(_("IP Address error"))
|
||||||
if ipa.is_private:
|
if ipa.is_private:
|
||||||
raise IpAddresValueError("Private IP Address not allowed")
|
raise IpAddresValueError("Private IP Address not allowed")
|
||||||
|
|
||||||
return True
|
|
||||||
|
|
|
@ -54,7 +54,8 @@ def test_webhook_action_test_transform_to_json(client, data):
|
||||||
response = Mock(status_code=200, headers={}, text="ok")
|
response = Mock(status_code=200, headers={}, text="ok")
|
||||||
response.elapsed.total_seconds.return_value = 100
|
response.elapsed.total_seconds.return_value = 100
|
||||||
|
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response), \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
client.login(data.project_owner)
|
client.login(data.project_owner)
|
||||||
response = client.json.post(url)
|
response = client.json.post(url)
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
|
|
|
@ -44,22 +44,26 @@ def test_new_object_with_one_webhook_signal(settings):
|
||||||
response.elapsed.total_seconds.return_value = 100
|
response.elapsed.total_seconds.return_value = 100
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test")
|
services.take_snapshot(obj, user=obj.owner, comment="test")
|
||||||
assert session_send_mock.call_count == 1
|
assert session_send_mock.call_count == 1
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner)
|
services.take_snapshot(obj, user=obj.owner)
|
||||||
assert session_send_mock.call_count == 0
|
assert session_send_mock.call_count == 0
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test")
|
services.take_snapshot(obj, user=obj.owner, comment="test")
|
||||||
assert session_send_mock.call_count == 1
|
assert session_send_mock.call_count == 1
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
||||||
assert session_send_mock.call_count == 1
|
assert session_send_mock.call_count == 1
|
||||||
|
|
||||||
|
@ -81,22 +85,26 @@ def test_new_object_with_two_webhook_signals(settings):
|
||||||
response.elapsed.total_seconds.return_value = 100
|
response.elapsed.total_seconds.return_value = 100
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test")
|
services.take_snapshot(obj, user=obj.owner, comment="test")
|
||||||
assert session_send_mock.call_count == 2
|
assert session_send_mock.call_count == 2
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test")
|
services.take_snapshot(obj, user=obj.owner, comment="test")
|
||||||
assert session_send_mock.call_count == 2
|
assert session_send_mock.call_count == 2
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner)
|
services.take_snapshot(obj, user=obj.owner)
|
||||||
assert session_send_mock.call_count == 0
|
assert session_send_mock.call_count == 0
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
||||||
assert session_send_mock.call_count == 2
|
assert session_send_mock.call_count == 2
|
||||||
|
|
||||||
|
@ -117,11 +125,13 @@ def test_send_request_one_webhook_signal(settings):
|
||||||
response.elapsed.total_seconds.return_value = 100
|
response.elapsed.total_seconds.return_value = 100
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test")
|
services.take_snapshot(obj, user=obj.owner, comment="test")
|
||||||
assert session_send_mock.call_count == 1
|
assert session_send_mock.call_count == 1
|
||||||
|
|
||||||
for obj in objects:
|
for obj in objects:
|
||||||
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
|
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \
|
||||||
|
patch("taiga.base.utils.urls.validate_destination_address", return_value=True):
|
||||||
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
|
||||||
assert session_send_mock.call_count == 1
|
assert session_send_mock.call_count == 1
|
||||||
|
|
|
@ -22,7 +22,8 @@ from unittest import mock
|
||||||
import django_sites as sites
|
import django_sites as sites
|
||||||
import re
|
import re
|
||||||
|
|
||||||
from taiga.base.utils.urls import get_absolute_url, is_absolute_url, build_url
|
from taiga.base.utils.urls import get_absolute_url, is_absolute_url, build_url, \
|
||||||
|
validate_private_url, IpAddresValueError
|
||||||
from taiga.base.utils.db import save_in_bulk, update_in_bulk, to_tsquery
|
from taiga.base.utils.db import save_in_bulk, update_in_bulk, to_tsquery
|
||||||
|
|
||||||
pytestmark = pytest.mark.django_db
|
pytestmark = pytest.mark.django_db
|
||||||
|
@ -124,7 +125,7 @@ def test_to_tsquery():
|
||||||
])
|
])
|
||||||
def test_validate_bad_destination_address(url):
|
def test_validate_bad_destination_address(url):
|
||||||
with pytest.raises(IpAddresValueError):
|
with pytest.raises(IpAddresValueError):
|
||||||
validate_destination_address(url)
|
validate_private_url(url)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("url", [
|
@pytest.mark.parametrize("url", [
|
||||||
|
@ -138,4 +139,4 @@ def test_validate_bad_destination_address(url):
|
||||||
"http://1.1.1.1/",
|
"http://1.1.1.1/",
|
||||||
])
|
])
|
||||||
def test_validate_good_destination_address(url):
|
def test_validate_good_destination_address(url):
|
||||||
assert validate_destination_address(url)
|
assert validate_private_url(url) is None
|
||||||
|
|
Loading…
Reference in New Issue