Revision ccdd1b82
b/kamaki/cli/commands/network.py | ||
---|---|---|
436 | 436 |
self._run(port_id=port_id) |
437 | 437 |
|
438 | 438 |
|
439 |
#@command(port_cmds) |
|
440 |
#class port_create(_init_network, _optional_json): |
|
441 |
# |
|
439 |
@command(port_cmds) |
|
440 |
class port_create(_init_network, _optional_json): |
|
441 |
"""Create a new port""" |
|
442 |
|
|
443 |
arguments = dict( |
|
444 |
security_group_id=RepeatableArgument( |
|
445 |
'Add a security group id (can be repeated)', |
|
446 |
('-g', '--security-group')) |
|
447 |
) |
|
448 |
|
|
449 |
@errors.generic.all |
|
450 |
@errors.cyclades.connection |
|
451 |
@errors.cyclades.network_id |
|
452 |
def _run(self, network_id, device_id): |
|
453 |
r = self.client.create_port( |
|
454 |
network_id, device_id, security_groups=self['security_group_id']) |
|
455 |
self._print(r, self.print_dict) |
|
456 |
|
|
457 |
def main(self, network_id, device_id): |
|
458 |
super(self.__class__, self)._run() |
|
459 |
self._run(network_id=network_id, device_id=device_id) |
b/kamaki/clients/cyclades/__init__.py | ||
---|---|---|
525 | 525 |
req['network']['shared'] = bool(shared) |
526 | 526 |
r = self.networks_post(json_data=req, success=201) |
527 | 527 |
return r.json['network'] |
528 |
|
|
529 |
def create_port(self, network_id, device_id, security_groups=None): |
|
530 |
port = dict(network_id=network_id, device_id=device_id) |
|
531 |
if security_groups: |
|
532 |
port['security_groups'] = security_groups |
|
533 |
r = self.ports_post(json_data=dict(port=port), success=201) |
|
534 |
return r.json['port'] |
b/kamaki/clients/cyclades/test.py | ||
---|---|---|
277 | 277 |
expargs = dict(json_data=dict(network=req), success=201) |
278 | 278 |
self.assertEqual(networks_post.mock_calls[-1], call(**expargs)) |
279 | 279 |
|
280 |
@patch( |
|
281 |
'kamaki.clients.network.rest_api.NetworkRestClient.ports_post', |
|
282 |
return_value=FR) |
|
283 |
def test_create_port(self, ports_post): |
|
284 |
network_id, device_id, FR.json = 'netid', 'devid', dict(port='ret v') |
|
285 |
for security_groups in ([1, 2, 3], None): |
|
286 |
self.assertEqual( |
|
287 |
self.client.create_port( |
|
288 |
network_id, device_id, security_groups=security_groups), |
|
289 |
'ret v') |
|
290 |
req = dict(network_id=network_id, device_id=device_id) |
|
291 |
if security_groups: |
|
292 |
req['security_groups'] = security_groups |
|
293 |
expargs = dict(json_data=dict(port=req), success=201) |
|
294 |
self.assertEqual(ports_post.mock_calls[-1], call(**expargs)) |
|
295 |
|
|
280 | 296 |
|
281 | 297 |
class CycladesClient(TestCase): |
282 | 298 |
|
Also available in: Unified diff