Add new transition actions: Assign new IP and Replace IP (#2845)

src/ralph/deployment/deployment.py

@@ -150,6 +150,23 @@ def next_free_ip_choices(actions, objects):
         )
         for network in networks
     ]
+    return ips
+
+
+def next_free_ip_choices_wth_other_choice(actions, objects):
+    """
+    Generate choices with next free IP for each network common for every object.
+    If there is only one object in this transition, custom IP address could be
+    passed (OTHER opiton).
+
+    Args:
+        actions: Transition action list
+        objects: Django models objects
+
+    Returns:
+        list of tuples with next free IP choices
+    """
+    ips = next_free_ip_choices(actions, objects)
     # if there is only one object, allow for Other option typed by user
     if len(objects) == 1:
         ips += [(OTHER, _('Other'))]
@@ -386,6 +403,10 @@ def _assign_hostname(instance, new_hostname, net_env=None):
         kwargs['history_kwargs'][instance.pk]['hostname'] = '{}{}'.format(
             new_hostname, ' (from {})'.format(net_env) if net_env else ''
         )
+        kwargs['shared_params']['hostnames'][instance.pk] = new_hostname
+
+    if 'hostnames' not in kwargs['shared_params']:
+        kwargs['shared_params']['hostnames'] = {}
 
     if network_environment['value'] == OTHER:
         hostname = network_environment[OTHER]
@@ -437,16 +458,115 @@ def check_ip_from_defined_network(address):
         )
 
 
+@deployment_action(
+    verbose_name=_('Assign new IP'),
+    form_fields={
+        'network': {
+            'field': forms.ChoiceField(
+                label=_('IP Address')
+            ),
+            'choices': next_free_ip_choices,
+            'exclude_from_history': True
+        },
+    },
+    run_after=['assign_new_hostname'],
+)
+def assign_new_ip(cls, instances, network, **kwargs):
+    for instance in instances:
+        network = Network.objects.get(pk=network)
+        ip = network.issue_next_free_ip()
+        logger.info('Assigning {} to {}'.format(ip, instance))
+        ethernet = Ethernet.objects.create(base_object=instance)
+        logger.info('Bounding {} to {} ethernet'.format(ip, ethernet))
+        ip.ethernet = ethernet
+
+        try:
+            hostname = kwargs['shared_params']['hostnames'].get(instance.pk)
+            if hostname:
+                ip.hostname = hostname
+        except KeyError:
+            pass
+
+        ip.save()
+
+
+def base_object_ip_choices(actions, objects):
+    ipaddresses = []
+    for instance in objects:
+        for ip in instance.ipaddresses:
+            ipaddresses.append(
+                [ip.id, '{} - {}'.format(ip.address, ip.hostname)]
+            )
+    return ipaddresses
+
+
+def base_object_network_choices(actions, objects):
+    networks = []
+    for instance in objects:
+        for network in instance._get_available_networks():
+            networks.append(
+                [network.id, network]
+            )
+    return networks
+
+
+def check_number_of_instance(instances):
+    """
+    Verify, if number of asset is equal to 1.
+
+    Args:
+        instances: Django model object instances
+    Returns:
+        errors: Dict
+    """
+    errors = {}
+    if len(instances) > 1:
+        errors[instances[0]] = _('You can choose only one asset')
+
+    return errors
+
+
+@deployment_action(
+    verbose_name=_('Replace IP'),
+    form_fields={
+        'ipaddress': {
+            'field': forms.ChoiceField(
+                label=_('IP Address'),
+            ),
+            'choices': base_object_ip_choices,
+            'exclude_from_history': True,
+        },
+        'network': {
+            'field': forms.ChoiceField(
+                label=_('Network'),
+            ),
+            'choices': base_object_network_choices,
+            'exclude_from_history': True,
+        },
+    },
+    precondition=check_number_of_instance
+)
+def replace_ip(cls, instances, ipaddress, network, **kwargs):
+    ip = IPAddress.objects.get(pk=ipaddress)
+    network = Network.objects.get(pk=network)
+    new_ip_address = str(network.get_first_free_ip())
+    logger.info(
+        'Replacing IP {} to {}'.format(ip.address, new_ip_address)
+    )
+    ip.address = new_ip_address
+    ip.save()
+
+
 @deployment_action(
     verbose_name=_('Assign new IP address and create DHCP entries'),
     form_fields={
         'ip_or_network': {
             'field': ChoiceFieldWithOtherOption(
                 label=_('IP Address'),
                 other_field=forms.GenericIPAddressField(),
-                auto_other_choice=False,
+                auto_other_choice=False
             ),
-            'choices': next_free_ip_choices,
+            'choices': next_free_ip_choices_wth_other_choice,
             'exclude_from_history': True,
             'validation': validate_ip_address,
         },

src/ralph/deployment/tests/test_deployment.py

@@ -178,7 +178,9 @@ def test_assign_new_hostname(self):
         next_free_hostname = 'server_10001.mydc.net'
         history = {self.instance.pk: {}}
         self.instance.__class__.assign_new_hostname(
-            [self.instance], {'value': self.net_env.id}, history_kwargs=history
+            [self.instance], {'value': self.net_env.id},
+            history_kwargs=history,
+            shared_params={'hostnames': {self.instance.pk: ''}}
         )
         self.assertEqual(self.instance.hostname, next_free_hostname)
         self.assertEqual(history, {
@@ -195,7 +197,8 @@ def test_assign_new_hostname_custom_value(self):
         self.instance.__class__.assign_new_hostname(
             [self.instance],
             {'value': '__other__', '__other__': 's12345.mydc.net'},
-            history_kwargs=history
+            history_kwargs=history,
+            shared_params={'hostnames': {self.instance.pk: ''}}
         )
         self.assertEqual(self.instance.hostname, 's12345.mydc.net')
         self.assertEqual(history, {

src/ralph/deployment/tests/test_transitions.py

@@ -1,9 +1,9 @@
 from ddt import data, ddt, unpack
+
 from django.contrib.auth import get_user_model
 from django.core.urlresolvers import reverse
 from django.test import Client
 from rest_framework.test import APIClient
-
 from ralph.assets.models import AssetLastHostname
 from ralph.assets.tests.factories import EthernetFactory
 from ralph.data_center.models import DataCenterAsset
@@ -92,6 +92,36 @@ def _prepare_assign_new_hostname_transition(self):
             target=str(TRANSITION_ORIGINAL_STATUS[0])
         )[1]
 
+    def _prepare_assign_new_ip_transition(self):
+        actions = ['assign_new_ip']
+        self.assign_assign_new_ip_transition = self._create_transition(
+            self.model, 'assign new ip', actions, 'status',
+            source=list(dict(
+                self.model._meta.get_field('status').choices
+            ).keys()),
+            target=str(TRANSITION_ORIGINAL_STATUS[0])
+        )[1]
+
+    def _prepare_assign_new_ip_with_dns_transition(self):
+        actions = ['assign_new_ip', 'assign_new_hostname']
+        self.assign_assign_new_ip_with_dns_transition = self._create_transition(
+            self.model, 'assign new ip with dns', actions, 'status',
+            source=list(dict(
+                self.model._meta.get_field('status').choices
+            ).keys()),
+            target=str(TRANSITION_ORIGINAL_STATUS[0])
+        )[1]
+
+    def _prepare_replace_ip_transition(self):
+        actions = ['replace_ip']
+        self.assign_replace_ip_transition = self._create_transition(
+            self.model, 'replace ip', actions, 'status',
+            source=list(dict(
+                self.model._meta.get_field('status').choices
+            ).keys()),
+            target=str(TRANSITION_ORIGINAL_STATUS[0])
+        )[1]
+
     def _get_transition_view_url(
         self, url_name, instance_id, transition_id=None, transition_name=None,
         app_label=None, model=None
@@ -276,6 +306,98 @@ def test_assign_new_hostname_through_gui(self):
         self.instance.refresh_from_db()
         self.assertEqual(self.instance.hostname, 'server_100012.mydc.net')
 
+    def test_assign_new_ip_through_gui(self):
+        self._prepare_assign_new_ip_transition()
+        network = self.instance._get_available_networks()[0]
+        ip = str(network.get_first_free_ip())
+        response = self.gui_client.post(
+            reverse(
+                self.transition_url_name,
+                args=(
+                    self.instance.id,
+                    self.assign_assign_new_ip_transition.id
+                )
+            ),
+            {
+                'assign_new_ip__network': network.id
+            },
+            follow=True
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(
+            response.redirect_chain[0][0],
+            reverse(self.redirect_url_name, args=(self.instance.id,))
+        )
+        self.instance.refresh_from_db()
+        self.assertTrue(
+            self.instance.ipaddresses.filter(
+                address=ip
+            ).exists()
+        )
+
+    def test_assign_new_ip_with_dns_through_gui(self):
+        self._prepare_assign_new_ip_with_dns_transition()
+        network = self.instance._get_available_networks()[0]
+        ip = str(network.get_first_free_ip())
+        response = self.gui_client.post(
+            reverse(
+                self.transition_url_name,
+                args=(
+                    self.instance.id,
+                    self.assign_assign_new_ip_with_dns_transition.id
+                )
+            ),
+            {
+                'assign_new_hostname__network_environment': '__other__',
+                'assign_new_hostname__network_environment__other__': 'hostname',  # noqa
+                'assign_new_ip__network': network.id,
+            },
+            follow=True
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(
+            response.redirect_chain[0][0],
+            reverse(self.redirect_url_name, args=(self.instance.id,))
+        )
+        self.instance.refresh_from_db()
+        self.assertTrue(
+            self.instance.ipaddresses.filter(
+                address=ip
+            ).exists()
+        )
+        self.assertEqual(self.instance.hostname, 'hostname')
+
+    def test_replace_ip_through_gui(self):
+        self._prepare_replace_ip_transition()
+        ethernet = EthernetFactory(base_object=self.instance)
+        ipaddress = IPAddressFactory(
+            address='10.20.30.1',
+            ethernet=ethernet,
+            hostname='test_hostname'
+        )
+        network = self.instance._get_available_networks()[0]
+        response = self.gui_client.post(
+            reverse(
+                self.transition_url_name,
+                args=(
+                    self.instance.id,
+                    self.assign_replace_ip_transition.id
+                )
+            ),
+            {
+                'replace_ip__ipaddress': ipaddress.id,
+                'replace_ip__network': network.id
+            },
+            follow=True
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(
+            response.redirect_chain[0][0],
+            reverse(self.redirect_url_name, args=(self.instance.id,))
+        )
+        ipaddress.refresh_from_db()
+        self.assertNotEqual(ipaddress.address, '10.20.30.1')
+
     def test_assign_new_hostname_through_gui_with_other_value(self):
         self._prepare_assign_new_hostname_transition()
         response = self.gui_client.post(