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
|
import json
from urllib.parse import unquote_plus
from fledge.tentative.resources import fledge_http_server_util
# Script to generate trusted bidding signals. The response depends on the
# keys and interestGroupNames - some result in entire response failures, others
# affect only their own value. Keys are preferentially used over
# interestGroupName, since keys are composible, but some tests need to cover
# there being no keys.
def main(request, response):
hostname = None
keys = None
interestGroupNames = None
# Manually parse query params. Can't use request.GET because it unescapes as well as splitting,
# and commas mean very different things from escaped commas.
for param in request.url_parts.query.split("&"):
pair = param.split("=", 1)
if len(pair) != 2:
return fail(response, "Bad query parameter: " + param)
# Browsers should escape query params consistently.
if "%20" in pair[1]:
return fail(response, "Query parameter should escape using '+': " + param)
# Hostname can't be empty. The empty string can be a key or interest group name, though.
if pair[0] == "hostname" and hostname == None and len(pair[1]) > 0:
hostname = pair[1]
continue
if pair[0] == "keys" and keys == None:
keys = list(map(unquote_plus, pair[1].split(",")))
continue
if pair[0] == "interestGroupNames" and interestGroupNames == None:
interestGroupNames = list(map(unquote_plus, pair[1].split(",")))
continue
if pair[0] == "slotSize" or pair[0] == "allSlotsRequestedSizes":
continue
return fail(response, "Unexpected query parameter: " + param)
# "interestGroupNames" and "hostname" are mandatory.
if not hostname:
return fail(response, "hostname missing")
if not interestGroupNames:
return fail(response, "interestGroupNames missing")
response.status = (200, b"OK")
# The JSON representation of this is used as the response body. This does
# not currently include a "perInterestGroupData" object.
responseBody = {"keys": {}}
# Set when certain special keys are observed, used in place of the JSON
# representation of `responseBody`, when set.
body = None
contentType = "application/json"
adAuctionAllowed = "true"
dataVersion = None
if keys:
for key in keys:
value = "default value"
if key == "close-connection":
# Close connection without writing anything, to simulate a
# network error. The write call is needed to avoid writing the
# default headers.
response.writer.write("")
response.close_connection = True
return
elif key.startswith("replace-body:"):
# Replace entire response body. Continue to run through other
# keys, to allow them to modify request headers.
body = key.split(':', 1)[1]
elif key.startswith("data-version:"):
dataVersion = key.split(':', 1)[1]
elif key == "http-error":
response.status = (404, b"Not found")
elif key == "no-content-type":
contentType = None
elif key == "wrong-content-type":
contentType = 'text/plain'
elif key == "bad-ad-auction-allowed":
adAuctionAllowed = "sometimes"
elif key == "ad-auction-not-allowed":
adAuctionAllowed = "false"
elif key == "no-ad-auction-allow":
adAuctionAllowed = None
elif key == "no-value":
continue
elif key == "wrong-value":
responseBody["keys"]["another-value"] = "another-value"
continue
elif key == "null-value":
value = None
elif key == "num-value":
value = 1
elif key == "string-value":
value = "1"
elif key == "array-value":
value = [1, "foo", None]
elif key == "object-value":
value = {"a":"b", "c":["d"]}
elif key == "interest-group-names":
value = json.dumps(interestGroupNames)
elif key == "hostname":
value = request.GET.first(b"hostname", b"not-found").decode("ASCII")
elif key == "headers":
value = fledge_http_server_util.headers_to_ascii(request.headers)
elif key == "slotSize":
value = request.GET.first(b"slotSize", b"not-found").decode("ASCII")
elif key == "allSlotsRequestedSizes":
value = request.GET.first(b"allSlotsRequestedSizes", b"not-found").decode("ASCII")
elif key == "url":
value = request.url
responseBody["keys"][key] = value
if "data-version" in interestGroupNames:
dataVersion = "4"
if contentType:
response.headers.set("Content-Type", contentType)
if adAuctionAllowed:
response.headers.set("Ad-Auction-Allowed", adAuctionAllowed)
if dataVersion:
response.headers.set("Data-Version", dataVersion)
response.headers.set("Ad-Auction-Bidding-Signals-Format-Version", "2")
if body != None:
return body
return json.dumps(responseBody)
def fail(response, body):
response.status = (400, "Bad Request")
response.headers.set(b"Content-Type", b"text/plain")
return body
|
