Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
itminedu
kamaki
Commits
7a3c66e1
Commit
7a3c66e1
authored
Dec 10, 2013
by
Stavros Sachtouris
Browse files
Fix unittests in kamaki.clients, subnet arg bug
parent
1d5653d9
Changes
9
Hide whitespace changes
Inline
Side-by-side
docs/commands.rst
View file @
7a3c66e1
...
...
@@ -8,9 +8,8 @@ Kamaki commands follow this scheme::
In this context, objects are not services, but virtual objects like a server, a
file or an image. The action concerns objects of the specified type. Some
actions (e.g. "delete" or "info") need to operate on an existing object. The
identifiers strictly identify this object and they can have the form of an id
(e.g., `server delete <SERVER_ID>`) or a foreign key (e.g.,
`port create <NETWORK_ID> <DEVICE_ID>`)
identifiers strictly identify this object and they should have the form of an id
(e.g., `server delete <SERVER_ID>`).
The examples bellow showcase some commands. The kamaki-shell (check
`Usage section <usage.html#interactive-shell>`_ for details) is chosen as the
...
...
@@ -351,7 +350,7 @@ Showcase: Connect a network to a VM
* Try network-connect (to get help) *
[network]: connect
Syntax error
usage: connect <network id>
<
device
id> [-s] [-h] [-i] [--config CONFIG]
usage: connect <network id>
--
device
-
id
<DEVICE_ID
> [-s] [-h] [-i] [--config CONFIG]
Connect a server to a network
...
...
@@ -364,7 +363,7 @@ Showcase: Connect a network to a VM
-v,--verbose: More info at response
* Connect VM with id 11687 to network with id 1409
[network]: connect 11687 1409 --wait
[network]: connect 11687
--device-id=
1409 --wait
Creating port between network 1409 and server 11687
New port: 8
...
...
kamaki/cli/commands/network.py
View file @
7a3c66e1
...
...
@@ -55,8 +55,9 @@ _commands = [network_cmds, port_cmds, subnet_cmds, ip_cmds]
about_authentication
=
'
\n
User Authentication:
\
\n
* to check authentication: /user authenticate
\
\n
* to set authentication token: /config set cloud.<cloud>.token <token>'
\n
to check authentication: [kamaki] ]user authenticate
\
\n
to set authentication token:
\
[kamaki] config set cloud.<CLOUD>.token <TOKEN>'
class
_port_wait
(
_service_wait
):
...
...
@@ -292,6 +293,8 @@ class AllocationPoolArgument(RepeatableArgument):
@
value
.
setter
def
value
(
self
,
new_pools
):
if
not
new_pools
:
return
new_list
=
[]
for
pool
in
new_pools
:
start
,
comma
,
end
=
pool
.
partition
(
','
)
...
...
@@ -635,7 +638,11 @@ class network_connect(_port_create):
ip_address
=
ValueArgument
(
'IP address for subnet id (used with --subnet-id'
,
'--ip-address'
),
wait
=
FlagArgument
(
'Wait network to connect'
,
(
'-w'
,
'--wait'
)),
device_id
=
RepeatableArgument
(
'Connect this device to the network (can be repeated)'
,
'--device-id'
)
)
required
=
(
'device_id'
,
)
@
errors
.
generic
.
all
@
errors
.
cyclades
.
connection
...
...
@@ -646,9 +653,10 @@ class network_connect(_port_create):
network_id
,
server_id
))
self
.
connect
(
network_id
,
server_id
)
def
main
(
self
,
network_id
,
device_id
):
def
main
(
self
,
network_id
):
super
(
self
.
__class__
,
self
).
_run
()
self
.
_run
(
network_id
=
network_id
,
server_id
=
device_id
)
for
sid
in
self
[
'device_id'
]:
self
.
_run
(
network_id
=
network_id
,
server_id
=
sid
)
@
command
(
network_cmds
)
...
...
@@ -663,8 +671,12 @@ class network_disconnect(_init_network, _port_wait, _optional_json):
return
CycladesClient
(
URL
,
self
.
client
.
token
)
arguments
=
dict
(
wait
=
FlagArgument
(
'Wait network to disconnect'
,
(
'-w'
,
'--wait'
))
wait
=
FlagArgument
(
'Wait network to disconnect'
,
(
'-w'
,
'--wait'
)),
device_id
=
RepeatableArgument
(
'Disconnect device from the network (can be repeated)'
,
'--device-id'
)
)
required
=
(
'device_id'
,
)
@
errors
.
generic
.
all
@
errors
.
cyclades
.
connection
...
...
@@ -673,7 +685,7 @@ class network_disconnect(_init_network, _port_wait, _optional_json):
def
_run
(
self
,
network_id
,
server_id
):
vm
=
self
.
_cyclades_client
().
get_server_details
(
server_id
)
ports
=
[
port
for
port
in
vm
[
'attachments'
]
if
(
port
[
'network_id'
]
not
in
(
'
network_id
'
,
))]
port
[
'network_id'
]
in
(
network_id
,
))]
if
not
ports
:
raiseCLIError
(
'Network %s is not connected to device %s'
%
(
network_id
,
server_id
))
...
...
@@ -692,6 +704,7 @@ class network_disconnect(_init_network, _port_wait, _optional_json):
raise
self
.
error
(
'Port %s is deleted'
%
port
[
'id'
])
def
main
(
self
,
network_id
,
device_id
):
def
main
(
self
,
network_id
):
super
(
self
.
__class__
,
self
).
_run
()
self
.
_run
(
network_id
=
network_id
,
server_id
=
device_id
)
for
sid
in
self
[
'device_id'
]:
self
.
_run
(
network_id
=
network_id
,
server_id
=
sid
)
kamaki/cli/commands/pithos.py
View file @
7a3c66e1
...
...
@@ -300,6 +300,12 @@ class file_list(_pithos_container, _optional_json, _name_filter):
if_unmodified_since
=
self
[
'if_unmodified_since'
],
until
=
self
[
'until'
],
meta
=
self
[
'meta'
])
# REMOVE THIS if version >> 0.12
if
not
r
.
json
:
self
.
error
(
' NOTE: Since v0.12, use / for containers e.g.,'
)
self
.
error
(
' [kamaki] file list /pithos'
)
files
=
self
.
_filter_by_name
(
r
.
json
)
if
self
[
'more'
]:
outbu
,
self
.
_out
=
self
.
_out
,
StringIO
()
...
...
kamaki/clients/cyclades/__init__.py
View file @
7a3c66e1
...
...
@@ -112,44 +112,6 @@ class CycladesClient(CycladesRestClient, Waiter):
r
=
self
.
servers_action_post
(
server_id
,
json_data
=
req
,
success
=
200
)
return
r
.
json
[
'console'
]
def
get_firewall_profile
(
self
,
server_id
):
"""
:param server_id: integer (str or int)
:returns: (str) ENABLED | DISABLED | PROTECTED
:raises ClientError: 520 No Firewall Profile
"""
r
=
self
.
get_server_details
(
server_id
)
try
:
return
r
[
'attachments'
][
0
][
'firewallProfile'
]
except
KeyError
:
raise
ClientError
(
'No Firewall Profile'
,
details
=
'Server %s is missing a firewall profile'
%
server_id
)
def
set_firewall_profile
(
self
,
server_id
,
profile
):
"""Set the firewall profile for the public interface of a server
:param server_id: integer (str or int)
:param profile: (str) ENABLED | DISABLED | PROTECTED
:returns: (dict) response headers
"""
req
=
{
'firewallProfile'
:
{
'profile'
:
profile
}}
r
=
self
.
servers_action_post
(
server_id
,
json_data
=
req
,
success
=
202
)
return
r
.
headers
def
list_server_nics
(
self
,
server_id
):
"""
:param server_id: integer (str or int)
:returns: (dict) network interface connections
"""
r
=
self
.
servers_ips_get
(
server_id
)
return
r
.
json
[
'attachments'
]
def
get_server_stats
(
self
,
server_id
):
"""
:param server_id: integer (str or int)
...
...
@@ -186,30 +148,6 @@ class CycladesClient(CycladesRestClient, Waiter):
return
self
.
_wait
(
server_id
,
current_status
,
get_status
,
delay
,
max_wait
,
wait_cb
)
def
wait_firewall
(
self
,
server_id
,
current_status
=
'DISABLED'
,
delay
=
1
,
max_wait
=
100
,
wait_cb
=
None
):
"""Wait while the public network firewall status is current_status
:param server_id: integer (str or int)
:param current_status: (str) DISABLED | ENABLED | PROTECTED
:param delay: time interval between retries
:max_wait: (int) timeout in secconds
:param wait_cb: if set a progressbar is used to show progress
:returns: (str) the new mode if succesfull, (bool) False if timed out
"""
def
get_status
(
self
,
server_id
):
return
self
.
get_firewall_profile
(
server_id
),
None
return
self
.
_wait
(
server_id
,
current_status
,
get_status
,
delay
,
max_wait
,
wait_cb
)
class
CycladesNetworkClient
(
NetworkClient
):
"""Cyclades Network API extentions"""
...
...
@@ -252,9 +190,7 @@ class CycladesNetworkClient(NetworkClient):
if
fixed_ips
:
for
fixed_ip
in
fixed_ips
or
[]:
if
not
'ip_address'
in
fixed_ip
:
raise
ValueError
(
'Invalid format for "fixed_ips"'
,
details
=
[
'fixed_ips format: [{"ip_address": IPv4}, ...]'
])
raise
ValueError
(
'Invalid fixed_ip [%s]'
%
fixed_ip
)
port
[
'fixed_ips'
]
=
fixed_ips
r
=
self
.
ports_post
(
json_data
=
dict
(
port
=
port
),
success
=
201
)
return
r
.
json
[
'port'
]
...
...
kamaki/clients/cyclades/rest_api.py
View file @
7a3c66e1
...
...
@@ -33,115 +33,12 @@
from
kamaki.clients.compute
import
ComputeClient
from
kamaki.clients.utils
import
path4url
import
json
class
CycladesRestClient
(
ComputeClient
):
"""Synnefo Cyclades REST API Client"""
def
servers_stats_get
(
self
,
server_id
,
success
=
200
,
**
kwargs
):
def
servers_stats_get
(
self
,
server_id
,
**
kwargs
):
"""GET base_url/servers/<server_id>/stats"""
path
=
path4url
(
'servers'
,
server_id
,
'stats'
)
return
self
.
get
(
path
,
success
=
success
,
**
kwargs
)
# def networks_get(
# self,
# network_id='',
# command='',
# success=(200, 203),
# **kwargs):
# """GET base_url/networks[/network_id][/command] request
# :param network_id: integer (str or int)
# :param command: (str) 'detail' or ''
# :param success: success code or list or tuple of accepted success
# codes. if server response code is not in this list, a ClientError
# raises
# :returns: request response
# """
# path = path4url('networks', network_id, command)
# return self.get(path, success=success, **kwargs)
# def networks_delete(
# self,
# network_id='',
# command='',
# success=204,
# **kwargs):
# """DEL ETE base_url/networks[/network_id][/command] request
# :param network_id: integer (str or int)
# :param command: (str) 'detail' or ''
# :param success: success code or list or tuple of accepted success
# codes. if server response code is not in this list, a ClientError
# raises
# :returns: request response
# """
# path = path4url('networks', network_id, command)
# return self.delete(path, success=success, **kwargs)
# def networks_post(
# self,
# network_id='',
# command='',
# json_data=None,
# success=202,
# **kwargs):
# """POST base_url/servers[/server_id]/[command] request
# :param network_id: integer (str or int)
# :param command: (str) 'detail' or ''
# :param json_data: (dict) will be send as data
# :param success: success code or list or tuple of accepted success
# codes. if server response code is not in this list, a ClientError
# raises
# :returns: request response
# """
# data = json_data
# if json_data is not None:
# data = json.dumps(json_data)
# self.set_header('Content-Type', 'application/json')
# self.set_header('Content-Length', len(data))
# path = path4url('networks', network_id, command)
# return self.post(path, data=data, success=success, **kwargs)
# def networks_put(
# self,
# network_id='',
# command='',
# json_data=None,
# success=204,
# **kwargs):
# """PUT base_url/servers[/server_id]/[command] request
# :param network_id: integer (str or int)
# :param command: (str) 'detail' or ''
# :param json_data: (dict) will be send as data
# :param success: success code or list or tuple of accepted success
# codes. if server response code is not in this list, a ClientError
# raises
# :returns: request response
# """
# data = json_data
# if json_data is not None:
# data = json.dumps(json_data)
# self.set_header('Content-Type', 'application/json')
# self.set_header('Content-Length', len(data))
# path = path4url('networks', network_id, command)
# return self.put(path, data=data, success=success, **kwargs)
return
self
.
get
(
path
,
success
=
200
,
**
kwargs
)
kamaki/clients/cyclades/test.py
View file @
7a3c66e1
...
...
@@ -94,149 +94,17 @@ cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
class
CycladesRestClient
(
TestCase
):
"""Set up a Cyclades thorough test"""
def
setUp
(
self
):
self
.
url
=
'http://cyclades.example.com'
self
.
token
=
'cyc14d3s70k3n'
self
.
client
=
cyclades
.
CycladesRestClient
(
self
.
url
,
self
.
token
)
def
tearDown
(
self
):
FR
.
json
=
vm_recv
@
patch
(
'%s.get'
%
rest_pkg
,
return_value
=
FR
())
def
test_networks_get
(
self
,
get
):
for
args
in
product
(
(
''
,
'net_id'
),
(
''
,
'cmd'
),
(
200
,
204
),
({},
{
'k'
:
'v'
})):
(
srv_id
,
command
,
success
,
kwargs
)
=
args
self
.
client
.
networks_get
(
*
args
[:
3
],
**
kwargs
)
srv_str
=
'/%s'
%
srv_id
if
srv_id
else
''
cmd_str
=
'/%s'
%
command
if
command
else
''
self
.
assertEqual
(
get
.
mock_calls
[
-
1
],
call
(
'/networks%s%s'
%
(
srv_str
,
cmd_str
),
success
=
success
,
**
kwargs
))
@
patch
(
'%s.delete'
%
rest_pkg
,
return_value
=
FR
())
def
test_networks_delete
(
self
,
delete
):
for
args
in
product
(
(
''
,
'net_id'
),
(
''
,
'cmd'
),
(
202
,
204
),
({},
{
'k'
:
'v'
})):
(
srv_id
,
command
,
success
,
kwargs
)
=
args
self
.
client
.
networks_delete
(
*
args
[:
3
],
**
kwargs
)
srv_str
=
'/%s'
%
srv_id
if
srv_id
else
''
cmd_str
=
'/%s'
%
command
if
command
else
''
self
.
assertEqual
(
delete
.
mock_calls
[
-
1
],
call
(
'/networks%s%s'
%
(
srv_str
,
cmd_str
),
success
=
success
,
**
kwargs
))
@
patch
(
'%s.set_header'
%
rest_pkg
)
@
patch
(
'%s.post'
%
rest_pkg
,
return_value
=
FR
())
def
test_networks_post
(
self
,
post
,
SH
):
for
args
in
product
(
(
''
,
'net_id'
),
(
''
,
'cmd'
),
(
None
,
[
dict
(
json
=
"data"
),
dict
(
data
=
"json"
)]),
(
202
,
204
),
({},
{
'k'
:
'v'
})):
(
srv_id
,
command
,
json_data
,
success
,
kwargs
)
=
args
self
.
client
.
networks_post
(
*
args
[:
4
],
**
kwargs
)
vm_str
=
'/%s'
%
srv_id
if
srv_id
else
''
cmd_str
=
'/%s'
%
command
if
command
else
''
if
json_data
:
json_data
=
dumps
(
json_data
)
self
.
assertEqual
(
SH
.
mock_calls
[
-
2
:],
[
call
(
'Content-Type'
,
'application/json'
),
call
(
'Content-Length'
,
len
(
json_data
))])
self
.
assertEqual
(
post
.
mock_calls
[
-
1
],
call
(
'/networks%s%s'
%
(
vm_str
,
cmd_str
),
data
=
json_data
,
success
=
success
,
**
kwargs
))
@
patch
(
'%s.set_header'
%
rest_pkg
)
@
patch
(
'%s.put'
%
rest_pkg
,
return_value
=
FR
())
def
test_networks_put
(
self
,
put
,
SH
):
for
args
in
product
(
(
''
,
'net_id'
),
(
''
,
'cmd'
),
(
None
,
[
dict
(
json
=
"data"
),
dict
(
data
=
"json"
)]),
(
202
,
204
),
({},
{
'k'
:
'v'
})):
(
srv_id
,
command
,
json_data
,
success
,
kwargs
)
=
args
self
.
client
.
networks_put
(
*
args
[:
4
],
**
kwargs
)
vm_str
=
'/%s'
%
srv_id
if
srv_id
else
''
cmd_str
=
'/%s'
%
command
if
command
else
''
if
json_data
:
json_data
=
dumps
(
json_data
)
self
.
assertEqual
(
SH
.
mock_calls
[
-
2
:],
[
call
(
'Content-Type'
,
'application/json'
),
call
(
'Content-Length'
,
len
(
json_data
))])
self
.
assertEqual
(
put
.
mock_calls
[
-
1
],
call
(
'/networks%s%s'
%
(
vm_str
,
cmd_str
),
data
=
json_data
,
success
=
success
,
**
kwargs
))
@
patch
(
'%s.get'
%
rest_pkg
,
return_value
=
FR
())
def
test_floating_ip_pools_get
(
self
,
get
):
for
args
in
product
(
(
200
,
204
),
({},
{
'k'
:
'v'
})):
success
,
kwargs
=
args
r
=
self
.
client
.
floating_ip_pools_get
(
success
,
**
kwargs
)
self
.
assertTrue
(
isinstance
(
r
,
FR
))
self
.
assertEqual
(
get
.
mock_calls
[
-
1
],
call
(
'/os-floating-ip-pools'
,
success
=
success
,
**
kwargs
))
@
patch
(
'%s.get'
%
rest_pkg
,
return_value
=
FR
())
def
test_floating_ips_get
(
self
,
get
):
for
args
in
product
(
(
'fip'
,
''
),
(
200
,
204
),
({},
{
'k'
:
'v'
})):
fip
,
success
,
kwargs
=
args
r
=
self
.
client
.
floating_ips_get
(
fip
,
success
,
**
kwargs
)
self
.
assertTrue
(
isinstance
(
r
,
FR
))
expected
=
''
if
not
fip
else
'/%s'
%
fip
self
.
assertEqual
(
get
.
mock_calls
[
-
1
],
call
(
'/os-floating-ips%s'
%
expected
,
success
=
success
,
**
kwargs
))
@
patch
(
'%s.set_header'
%
rest_pkg
)
@
patch
(
'%s.post'
%
rest_pkg
,
return_value
=
FR
())
def
test_floating_ips_post
(
self
,
post
,
SH
):
for
args
in
product
(
(
None
,
[
dict
(
json
=
"data"
),
dict
(
data
=
"json"
)]),
(
'fip'
,
''
),
(
202
,
204
),
({},
{
'k'
:
'v'
})):
json_data
,
fip
,
success
,
kwargs
=
args
self
.
client
.
floating_ips_post
(
*
args
[:
3
],
**
kwargs
)
if
json_data
:
json_data
=
dumps
(
json_data
)
self
.
assertEqual
(
SH
.
mock_calls
[
-
2
:],
[
call
(
'Content-Type'
,
'application/json'
),
call
(
'Content-Length'
,
len
(
json_data
))])
expected
=
''
if
not
fip
else
'/%s'
%
fip
self
.
assertEqual
(
post
.
mock_calls
[
-
1
],
call
(
'/os-floating-ips%s'
%
expected
,
data
=
json_data
,
success
=
success
,
**
kwargs
))
@
patch
(
'%s.delete'
%
rest_pkg
,
return_value
=
FR
())
def
test_floating_ips_delete
(
self
,
delete
):
for
args
in
product
(
(
'fip1'
,
'fip2'
),
(
200
,
204
),
({},
{
'k'
:
'v'
})):
fip
,
success
,
kwargs
=
args
r
=
self
.
client
.
floating_ips_delete
(
fip
,
success
,
**
kwargs
)
self
.
assertTrue
(
isinstance
(
r
,
FR
))
self
.
assertEqual
(
delete
.
mock_calls
[
-
1
],
call
(
'/os-floating-ips/%s'
%
fip
,
success
=
success
,
**
kwargs
))
@
patch
(
'kamaki.clients.Client.get'
,
return_value
=
'ret'
)
def
test_servers_stats_get
(
self
,
get
):
server_id
=
'server id'
self
.
assertEqual
(
self
.
client
.
servers_stats_get
(
server_id
),
'ret'
)
get
.
assert_called_once_with
(
'/servers/%s/stats'
%
server_id
,
success
=
200
)
class
CycladesNetworkClient
(
TestCase
):
...
...
@@ -286,12 +154,12 @@ class CycladesNetworkClient(TestCase):
(
'port name'
,
None
),
([
1
,
2
,
3
],
None
),
(
dict
(
subnet_id
=
'sid'
,
ip_address
=
'ipa'
),
dict
(
subnet_id
=
'sid'
),
dict
(
ip_address
=
'ipa'
),
[
dict
(
subnet_id
=
'sid'
,
ip_address
=
'ipa'
)
]
,
[
dict
(
subnet_id
=
'sid'
)
]
,
[
dict
(
ip_address
=
'ipa'
)
]
,
None
)):
if
fixed_ips
:
diff
=
set
([
'subnet_id'
,
'ip_address'
]).
difference
(
fixed_ips
)
diff
=
set
([
'ip_address'
,
]).
difference
(
fixed_ips
[
0
]
)
if
diff
:
self
.
assertRaises
(
ValueError
,
self
.
client
.
create_port
,
...
...
@@ -360,249 +228,6 @@ class CycladesClient(TestCase):
vm_id
,
json_data
=
dict
(
console
=
dict
(
type
=
'vnc'
)),
success
=
200
)
self
.
assert_dicts_are_equal
(
r
,
cnsl
[
'console'
])
def
test_get_firewall_profile
(
self
):
vm_id
=
vm_recv
[
'server'
][
'id'
]
v
=
firewalls
[
'attachments'
][
0
][
'firewallProfile'
]
with
patch
.
object
(
cyclades
.
CycladesClient
,
'get_server_details'
,
return_value
=
firewalls
)
as
GSD
:
r
=
self
.
client
.
get_firewall_profile
(
vm_id
)
GSD
.
assert_called_once_with
(
vm_id
)
self
.
assertEqual
(
r
,
v
)
with
patch
.
object
(
cyclades
.
CycladesClient
,
'get_server_details'
,
return_value
=
dict
()):
self
.
assertRaises
(
ClientError
,
self
.
client
.
get_firewall_profile
,
vm_id
)
@
patch
(
'%s.servers_action_post'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_set_firewall_profile
(
self
,
SP
):
vm_id
=
vm_recv
[
'server'
][
'id'
]
v
=
firewalls
[
'attachments'
][
0
][
'firewallProfile'
]
self
.
client
.
set_firewall_profile
(
vm_id
,
v
)
SP
.
assert_called_once_with
(
vm_id
,
json_data
=
dict
(
firewallProfile
=
dict
(
profile
=
v
)),
success
=
202
)
@
patch
(
'%s.networks_post'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_create_network
(
self
,
NP
):
net_name
=
net_send
[
'network'
][
'name'
]
FR
.
json
=
net_recv
full_args
=
dict
(
cidr
=
'192.168.0.0/24'
,
gateway
=
'192.168.0.1'
,
type
=
'MAC_FILTERED'
,
dhcp
=
True
)
test_args
=
dict
(
full_args
)
test_args
.
update
(
dict
(
empty
=
None
,
full
=
None
))
net_exp
=
dict
(
dhcp
=
False
,
name
=
net_name
,
type
=
'MAC_FILTERED'
)
for
arg
,
val
in
test_args
.
items
():
kwargs
=
{}
if
arg
==
'empty'
else
full_args
if
(
arg
==
'full'
)
else
{
arg
:
val
}
expected
=
dict
(
network
=
dict
(
net_exp
))
expected
[
'network'
].
update
(
kwargs
)
r
=
self
.
client
.
create_network
(
net_name
,
**
kwargs
)
self
.
assertEqual
(
NP
.
mock_calls
[
-
1
],
call
(
json_data
=
expected
,
success
=
202
))
self
.
assert_dicts_are_equal
(
r
,
net_recv
[
'network'
])
@
patch
(
'%s.networks_post'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_connect_server
(
self
,
NP
):
vm_id
=
vm_recv
[
'server'
][
'id'
]
net_id
=
net_recv
[
'network'
][
'id'
]
self
.
client
.
connect_server
(
vm_id
,
net_id
)
NP
.
assert_called_once_with
(
net_id
,
'action'
,
json_data
=
dict
(
add
=
dict
(
serverRef
=
vm_id
)))
@
patch
(
'%s.networks_post'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_disconnect_server
(
self
,
NP
):
net_id
,
vm_id
=
net_recv
[
'network'
][
'id'
],
vm_recv
[
'server'
][
'id'
]
nic_id
=
'nic-%s-%s'
%
(
net_id
,
vm_id
)
vm_nics
=
[
dict
(
id
=
nic_id
,
network_id
=
net_id
),
dict
(
id
=
'another-nic-id'
,
network_id
=
'another-net-id'
),
dict
(
id
=
nic_id
*
2
,
network_id
=
net_id
*
2
)]
with
patch
.
object
(
cyclades
.
CycladesClient
,
'list_server_nics'
,
return_value
=
vm_nics
)
as
LSN
:
r
=
self
.
client
.
disconnect_server
(
vm_id
,
nic_id
)
LSN
.
assert_called_once_with
(
vm_id
)
NP
.
assert_called_once_with
(
net_id
,
'action'
,
json_data
=
dict
(
remove
=
dict
(
attachment
=
nic_id
)))
self
.
assertEqual
(
r
,
1
)
@
patch
(
'%s.servers_ips_get'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_list_server_nics
(
self
,
SG
):
vm_id
=
vm_recv
[
'server'
][
'id'
]
nics
=
dict
(
attachments
=
[
dict
(
id
=
'nic1'
),
dict
(
id
=
'nic2'
)])
FR
.
json
=
nics
r
=
self
.
client
.
list_server_nics
(
vm_id
)
SG
.
assert_called_once_with
(
vm_id
)
expected
=
nics
[
'attachments'
]
for
i
in
range
(
len
(
r
)):
self
.
assert_dicts_are_equal
(
r
[
i
],
expected
[
i
])
self
.
assertEqual
(
i
+
1
,
len
(
r
))
@
patch
(
'%s.networks_get'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_list_networks
(
self
,
NG
):
FR
.
json
=
net_list
expected
=
net_list
[
'networks'
]
for
detail
in
(
''
,
'detail'
):
r
=
self
.
client
.
list_networks
(
detail
=
True
if
detail
else
False
)
self
.
assertEqual
(
NG
.
mock_calls
[
-
1
],
call
(
command
=
detail
))
for
i
,
net
in
enumerate
(
expected
):
self
.
assert_dicts_are_equal
(
r
[
i
],
net
)
self
.
assertEqual
(
i
+
1
,
len
(
r
))
@
patch
(
'%s.networks_get'
%
cyclades_pkg
,
return_value
=
FR
())
def
test_list_network_nics
(
self
,
NG
):
net_id
=
net_recv
[
'network'
][
'id'
]
FR
.
json
=
net_recv