1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# Copyright 2012-2015 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
from __future__ import (
absolute_import,
print_function,
unicode_literals,
)
"""Tests for `maasserver`."""
str = None
__metaclass__ = type
__all__ = [
"extract_redirect",
"get_content_links",
"get_data",
"get_prefixed_form_data",
"NoReceivers",
]
import collections
from contextlib import contextmanager
import httplib
from itertools import chain
import os
from urlparse import urlparse
import crochet
from lxml.html import fromstring
# Many tests need a running reactor.
crochet.setup()
def extract_redirect(http_response):
"""Extract redirect target from an http response object.
Only the http path part of the redirect is ignored; protocol and host
name, if present, are not included in the result.
If the response is not a redirect, this raises :class:`ValueError` with
a descriptive error message.
:param http_response: A response returned from an http request.
:type http_response: :class:`HttpResponse`
:return: The "path" part of the target that `http_response` redirects to.
:raises: ValueError
"""
if http_response.status_code != httplib.FOUND:
raise ValueError(
"Not a redirect: http status %d. Content: %s"
% (http_response.status_code, http_response.content[:80]))
target_url = http_response['Location']
parsed_url = urlparse(target_url)
return parsed_url.path
def get_data(filename):
"""Read the content of a file in `src/maasserver/tests`.
Some tests use this to read fixed data stored in files in
`src/maasserver/tests/data/`.
Where possible, provide data in-line in tests, or use fakes, to keep the
information close to the tests that rely on it.
:param filename: A file path relative to `src/maasserver/tests` in
this branch.
:return: Binary contents of the file, as `bytes`.
"""
path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), '..', 'tests', filename)
return file(path).read()
def get_prefixed_form_data(prefix, data):
"""Prefix entries in a dict of form parameters with a form prefix.
Also, add a parameter "<prefix>_submit" to indicate that the form with
the given prefix is being submitted.
Use this to construct a form submission if the form uses a prefix (as it
would if there are multiple forms on the page).
:param prefix: Form prefix string.
:param data: A dict of form parameters.
:return: A new dict of prefixed form parameters.
"""
result = {'%s-%s' % (prefix, key): value for key, value in data.items()}
result.update({'%s_submit' % prefix: 1})
return result
def get_content_links(response, element='#content'):
"""Extract links from :class:`HttpResponse` content.
:param response: An HTTP response object. Only its `content` attribute is
used.
:param element: Optional CSS selector for the node(s) in the content whose
links should be extracted. Only links inside the part of the content
that matches this selector will be extracted; any other links will be
ignored. Defaults to `#content`, which is the main document.
:return: List of link targets found in any matching parts of the document,
including their nested tags. If a link is in a DOM subtree that
matches `element` at multiple levels, it may be counted more than once.
Otherwise, links are returned in the same order in which they are found
in the document.
"""
doc = fromstring(response.content)
links_per_matching_node = chain.from_iterable(
[elem.get('href') for elem in matching_node.cssselect('a')]
for matching_node in doc.cssselect(element)
)
return list(links_per_matching_node)
@contextmanager
def NoReceivers(signals):
"""Disconnect signal receivers from the supplied signals.
:param signals: A signal (or iterable of signals) for which to disable
signal receivers while in the context manager.
:type signal: django.dispatch.Signal
"""
saved = dict()
if not isinstance(signals, collections.Iterable):
signals = [signals]
for signal in signals:
saved[signal] = signal.receivers
signal.receivers = []
try:
yield
finally:
for signal in signals:
signal.receivers = saved[signal]
|