Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
S
slapos.core
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Paul Graydon
slapos.core
Commits
4239a3a0
Commit
4239a3a0
authored
Aug 03, 2011
by
Antoine Catton
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Suppress Flack Web Server by patching httplib
parent
01abe672
Changes
3
Hide whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
149 additions
and
28 deletions
+149
-28
slapos/tests/mock/__init__.py
slapos/tests/mock/__init__.py
+0
-0
slapos/tests/mock/httplib.py
slapos/tests/mock/httplib.py
+115
-0
slapos/tests/slapgrid.py
slapos/tests/slapgrid.py
+34
-28
No files found.
slapos/tests/mock/__init__.py
0 → 100644
View file @
4239a3a0
slapos/tests/mock/httplib.py
0 → 100644
View file @
4239a3a0
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
"""Mocked httplib
"""
from
urlparse
import
urlparse
__all__
=
[]
def
log
(
message
):
"""Need to be overridden to get a proper logger
"""
pass
class
HTTPConnection
(
object
):
scheme
=
'http'
def
_callback
(
self
,
path
,
method
,
body
,
headers
):
"""To get it works properly, you need to override
HTTPConnection._callback. This method received the instance, the path,
method and request body as parameter, and it has to return a tuple with
headers dictionary and body response string.
@param self object instance reference
@param URL the parsed URL
@param method the http method
@param body the request body
@param headers the request headers
@return tuple containing status integer, headers dictionary and body
response"""
return
(
0
,
{},
''
,
)
def
__init__
(
self
,
host
,
port
=
None
,
strict
=
None
,
timeout
=
None
,
source_address
=
None
):
self
.
host
=
host
self
.
port
=
port
self
.
strict
=
strict
self
.
timeout
=
timeout
self
.
source_address
=
source_address
self
.
__response
=
None
def
request
(
self
,
method
,
url
,
body
=
None
,
headers
=
None
):
status
,
headers
,
body
=
self
.
_callback
(
url
,
method
,
body
,
headers
)
self
.
__response
=
HTTPResponse
(
'HTTP/1.1'
,
200
,
'OK'
,
body
,
headers
)
def
getresponse
(
self
):
response
=
self
.
__response
self
.
__response
=
None
return
response
def
set_debuglevel
(
self
,
level
):
pass
def
set_tunnel
(
self
,
host
,
port
=
None
,
headers
=
None
):
pass
def
connect
(
self
):
pass
def
close
(
self
):
pass
def
putrequest
(
self
,
request
,
selector
,
skip_host
=
None
,
skip_accept_encoding
=
None
):
pass
def
putheader
(
self
,
*
args
):
pass
def
endheaders
(
self
):
pass
def
send
(
self
,
data
):
pass
class
HTTPSConnection
(
HTTPConnection
):
def
__init__
(
self
,
host
,
port
=
None
,
key_file
=
None
,
cert_file
=
None
,
strict
=
None
,
timeout
=
None
,
source_address
=
None
):
super
().
__init__
(
self
,
host
,
port
,
strict
,
timeout
,
source_address
)
pass
class
HTTPResponse
(
object
):
def
__init__
(
self
,
version
,
status
,
reason
,
content
,
headers
=
()):
self
.
version
=
version
self
.
status
=
status
self
.
reason
=
reason
self
.
__headers
=
headers
self
.
__content
=
content
def
read
(
self
,
amt
=
None
):
result
=
None
if
amt
is
None
:
result
=
self
.
__content
self
.
__content
=
''
else
:
end
=
max
(
amt
,
len
(
self
.
__content
))
result
=
self
.
__content
[:
end
]
del
self
.
__content
[:
end
]
return
result
def
getheader
(
self
,
name
,
default
=
None
):
pass
def
getheaders
(
self
):
pass
class
HTTPMessage
(
object
):
pass
slapos/tests/slapgrid.py
View file @
4239a3a0
from
slapos.grid
import
slapgrid
import
flask
import
multiprocessing
import
os
import
shutil
import
signal
...
...
@@ -9,8 +7,7 @@ import socket
import
tempfile
import
unittest
import
xml_marshaller
app
=
flask
.
Flask
(
__name__
)
import
httplib
class
BasicMixin
:
def
setUp
(
self
):
...
...
@@ -46,31 +43,31 @@ class TestBasicSlapgridCP(BasicMixin, unittest.TestCase):
os
.
mkdir
(
self
.
instance_root
)
self
.
assertRaises
(
socket
.
error
,
self
.
grid
.
processComputerPartitionList
)
def
_run_server
(
host
,
port
):
global
app
app
.
run
(
host
=
host
,
port
=
port
,
use_reloader
=
False
,
debug
=
True
)
class
MasterMixin
(
BasicMixin
):
_master_port
=
45678
_master_host
=
'127.0.0.1'
def
startMaster
(
self
):
self
.
process
=
multiprocessing
.
Process
(
target
=
_run_server
,
args
=
(
self
.
_master_host
,
self
.
_master_port
))
self
.
process
.
start
()
self
.
master_url
=
'http://%s:%s/'
%
(
self
.
_master_host
,
self
.
_master_port
)
def
_patchHttplib
(
self
):
# XXX-Antoine: save and override the httplib
import
mock.httplib
self
.
saved_httplib
=
dict
()
for
fake
in
vars
(
mock
.
httplib
):
self
.
saved_httplib
[
fake
]
=
getattr
(
httplib
,
fake
,
None
)
setattr
(
httplib
,
fake
,
getattr
(
mock
.
httplib
,
fake
))
def
stopMaster
(
self
):
self
.
process
.
terminate
()
self
.
process
.
join
()
def
_unpatchHttplib
(
self
):
# XXX-Antoine: restore the httplib like it was
import
httplib
for
name
,
original_value
in
self
.
saved_httplib
.
items
():
setattr
(
httplib
,
name
,
original_value
)
del
self
.
saved_httplib
def
setUp
(
self
):
# prepare master
self
.
startMaster
()
self
.
_patchHttplib
()
BasicMixin
.
setUp
(
self
)
def
tearDown
(
self
):
self
.
stopMaster
()
self
.
_unpatchHttplib
()
# XXX: Hardcoded pid, as it is not configurable in slapos
svc
=
os
.
path
.
join
(
self
.
instance_root
,
'var'
,
'run'
,
'supervisord.pid'
)
if
os
.
path
.
exists
(
svc
):
...
...
@@ -83,15 +80,24 @@ class MasterMixin(BasicMixin):
BasicMixin
.
tearDown
(
self
)
class
TestSlapgridCPWithMaster
(
MasterMixin
,
unittest
.
TestCase
):
@
app
.
route
(
'/getComputerInformation'
,
methods
=
[
'GET'
])
def
getComputerInformation
():
computer_id
=
flask
.
request
.
args
[
'computer_id'
]
slap_computer
=
slapos
.
slap
.
Computer
(
computer_id
)
slap_computer
.
_software_release_list
=
[]
slap_computer
.
_computer_partition_list
=
[]
return
xml_marshaller
.
xml_marshaller
.
dumps
(
slap_computer
)
def
test_nothing_to_do
(
self
):
def
server_response
(
self
,
path
,
method
,
body
,
header
):
import
urlparse
parsed_url
=
urlparse
.
urlparse
(
'/'
+
path
)
parsed_qs
=
urlparse
.
parse_qs
(
parsed_url
.
query
)
if
parsed_url
.
path
==
'/getComputerInformation'
and
\
'computer_id'
in
parsed_qs
:
slap_computer
=
slapos
.
slap
.
Computer
(
parsed_qs
[
'computer_id'
])
slap_computer
.
_software_release_list
=
[]
slap_computer
.
_computer_partition_list
=
[]
return
(
200
,
{},
xml_marshaller
.
xml_marshaller
.
dumps
(
slap_computer
))
else
:
return
(
404
,
{},
''
)
httplib
.
HTTPConnection
.
_callback
=
server_response
os
.
mkdir
(
self
.
software_root
)
os
.
mkdir
(
self
.
instance_root
)
self
.
assertTrue
(
self
.
grid
.
processComputerPartitionList
())
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment