change endpoint live

ereuse_devicehub/resources/action/__init__.py

@@ -3,7 +3,8 @@ from typing import Callable, Iterable, Tuple
 from teal.resource import Converters, Resource
 
 from ereuse_devicehub.resources.action import schemas
-from ereuse_devicehub.resources.action.views import ActionView, AllocateView, DeallocateView
+from ereuse_devicehub.resources.action.views import (ActionView, AllocateView, DeallocateView,
+                                                     LiveView)
 from ereuse_devicehub.resources.device.sync import Sync
 
 
@@ -214,7 +215,7 @@ class PrepareDef(ActionDef):
 
 
 class LiveDef(ActionDef):
-    VIEW = None
+    VIEW = LiveView
     SCHEMA = schemas.Live
 
 

ereuse_devicehub/resources/action/models.py

@@ -1304,6 +1304,9 @@ class Live(JoinedWithOneDeviceMixin, ActionWithOneDevice):
     serial_number.comment = """The serial number of the Hard Disk in lower case."""
     usage_time_hdd = Column(Interval, nullable=True)
     snapshot_uuid = Column(UUID(as_uuid=True))
+    software = Column(DBEnum(SnapshotSoftware), nullable=False)
+    software_version = Column(StrictVersionType(STR_SM_SIZE), nullable=False)
+    licence_version = Column(StrictVersionType(STR_SM_SIZE), nullable=False)
 
     @property
     def final_user_code(self):
@@ -1347,6 +1350,7 @@ class Live(JoinedWithOneDeviceMixin, ActionWithOneDevice):
         return timedelta(0)
 
     def diff_time(self):
+        import pdb; pdb.set_trace()
         for e in self.actions:
             if e.created > self.created:
                 continue

ereuse_devicehub/resources/action/schemas.py

@@ -420,11 +420,30 @@ class Prepare(ActionWithMultipleDevices):
 
 class Live(ActionWithOneDevice):
     __doc__ = m.Live.__doc__
+    """
+    The Snapshot updates the state of the device with information about
+    its components and actions performed at them.
+
+    See docs for more info.
+    """
+    uuid = UUID()
+    software = EnumField(SnapshotSoftware,
+                         required=True,
+                         description='The software that generated this Snapshot.')
+    version = Version(required=True, description='The version of the software.')
     final_user_code = SanitizedStr(data_key="finalUserCode", dump_only=True)
-    serial_number = SanitizedStr(data_key="serialNumber", dump_only=True)
+    software = EnumField(SnapshotSoftware,
-    usage_time_hdd = TimeDelta(data_key="usageTimeHdd", precision=TimeDelta.HOURS, dump_only=True)
+                         required=True,
-    usage_time_allocate = TimeDelta(data_key="usageTimeAllocate", 
+                         description='The software that generated this Snapshot.')
-                                    precision=TimeDelta.HOURS, dump_only=True)
+    version = Version(required=True, description='The version of the software.')
+    licence_version = Version(required=True, description='The version of the software.')
+    components = NestedOn(s_device.Component,
+                          many=True,
+                          description='A list of components that are inside of the device'
+                                      'at the moment of this Snapshot.'
+                                      'Order is preserved, so the component num 0 when'
+                                      'submitting is the component num 0 when returning it back.')
+    usage_time_allocate = TimeDelta(data_key='usageTimeAllocate', required=False,)
 
 
 class Organize(ActionWithMultipleDevices):

ereuse_devicehub/resources/action/views.py

@@ -96,10 +96,116 @@ class AllocateMix():
 class AllocateView(AllocateMix, View):
     model = Allocate
 
+
 class DeallocateView(AllocateMix, View):
     model = Deallocate
 
 
+class LiveView(View):
+    def post(self):
+        """Posts an action."""
+        res_json = request.get_json(validate=False)
+        tmp_snapshots = app.config['TMP_SNAPSHOTS']
+        path_live = save_json(res_json, tmp_snapshots, g.user.email)
+        res_json_valid = request.get_json()
+        live = self.live(res_json_valid)
+        db.session.add(live)
+        db.session().final_flush()
+        ret = self.schema.jsonify(live)
+        ret.status_code = 201
+        db.session.commit()
+        move_json(tmp_snapshots, path_live, g.user.email)
+        return ret
+
+    def get_hdd_details(self, snapshot, device):
+        """We get the liftime and serial_number of the disk"""
+        usage_time_hdd = None
+        serial_number = None
+        for hd in snapshot['components']:
+            if not isinstance(hd, DataStorage):
+                continue
+
+            serial_number = hd.serial_number
+            for act in hd.actions:
+                if not act.type == "TestDataStorage":
+                    continue
+                usage_time_hdd = act.lifetime
+                break
+
+            if usage_time_hdd:
+                break
+
+        if not serial_number:
+            """There aren't any disk"""
+            raise ResourceNotFound("There aren't any disk in this device {}".format(device))
+        return usage_time_hdd, serial_number
+
+    def get_hid(self, snapshot):
+        device = snapshot.get('device')  # type: Computer
+        components = snapshot.get('components')
+        if not device:
+            return None
+        if not components:
+            return device.hid
+        macs = [c.serial_number for c in components
+                if c.type == 'NetworkAdapter' and c.serial_number is not None]
+        macs.sort()
+        mac = ''
+        hid = device.hid
+        if not hid:
+            return hid
+        if macs:
+            mac = "-{mac}".format(mac=macs[0])
+        hid += mac
+        return hid
+
+    def live(self, snapshot):
+        """If the device.allocated == True, then this snapshot create an action live."""
+        hid = self.get_hid(snapshot)
+        if not hid or not Device.query.filter(
+            Device.hid==hid, Device.owner_id==g.user.id).count():
+            raise ValidationError('Device not exist.')
+
+        device = Device.query.filter(
+            Device.hid==hid, Device.owner_id==g.user.id).one()
+        # Is not necessary
+        if not device:
+            raise ValidationError('Device not exist.')
+        if not device.allocated:
+            raise ValidationError('Sorry this device is not allocated.')
+
+        usage_time_hdd, serial_number = self.get_hdd_details(snapshot, device)
+
+        data_live = {'usage_time_hdd': usage_time_hdd,
+                     'serial_number': serial_number,
+                     'snapshot_uuid': snapshot['uuid'],
+                     'description': '',
+                     'software': snapshot['software'],
+                     'software_version': snapshot['version'],
+                     'licence_version': snapshot['licence_version'],
+                     'device': device}
+
+        live = Live(**data_live)
+
+        if not usage_time_hdd:
+            warning = f"We don't found any TestDataStorage for disk sn: {serial_number}"
+            live.severity = Severity.Warning
+            live.description = warning
+            return live
+
+        live.sort_actions()
+        diff_time = live.diff_time()
+        if diff_time is None:
+            warning = "Don't exist one previous live or snapshot as reference"
+            live.description += warning
+            live.severity = Severity.Warning
+        elif diff_time < timedelta(0):
+            warning = "The difference with the last live/snapshot is negative"
+            live.description += warning
+            live.severity = Severity.Warning
+        return live
+
+
 class ActionView(View):
     def post(self):
         """Posts an action."""
@@ -146,16 +252,6 @@ class ActionView(View):
         # model object, when we flush them to the db we will flush
         # snapshot, and we want to wait to flush snapshot at the end
 
-        # If the device is allocated, then snapshot is a live 
-        live = self.live(snapshot_json)
-        if live:
-            db.session.add(live)
-            db.session().final_flush()
-            ret = self.schema.jsonify(live)  # transform it back
-            ret.status_code = 201
-            db.session.commit()
-            return ret
-
         device = snapshot_json.pop('device')  # type: Computer
         components = None
         if snapshot_json['software'] == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid):
@@ -216,89 +312,6 @@ class ActionView(View):
         db.session.commit()
         return ret
 
-    def get_hdd_details(self, snapshot, device):
-        """We get the liftime and serial_number of the disk"""
-        usage_time_hdd = None
-        serial_number = None
-        for hd in snapshot['components']:
-            if not isinstance(hd, DataStorage):
-                continue
-
-            serial_number = hd.serial_number
-            for act in hd.actions:
-                if not act.type == "TestDataStorage":
-                    continue
-                usage_time_hdd = act.lifetime
-                break
-
-            if usage_time_hdd:
-                break
-
-        if not serial_number:
-            "There aren't any disk"
-            raise ResourceNotFound("There aren't any disk in this device {}".format(device))
-        return usage_time_hdd, serial_number
-
-    def get_hid(self, snapshot):
-        device = snapshot.get('device')  # type: Computer
-        components = snapshot.get('components')
-        if not device:
-            return None
-        if not components:
-            return device.hid
-        macs = [c.serial_number for c in components
-                if c.type == 'NetworkAdapter' and c.serial_number is not None]
-        macs.sort()
-        mac = ''
-        hid = device.hid
-        if not hid:
-            return hid
-        if macs:
-            mac = "-{mac}".format(mac=macs[0])
-        hid += mac
-        return hid
-
-    def live(self, snapshot):
-        """If the device.allocated == True, then this snapshot create an action live."""
-        hid = self.get_hid(snapshot)
-        if not hid or not Device.query.filter(
-            Device.hid==hid, Device.owner_id==g.user.id).count():
-            return None
-
-        device = Device.query.filter(
-            Device.hid==hid, Device.owner_id==g.user.id).one()
-
-        if not device.allocated:
-            return None
-
-        usage_time_hdd, serial_number = self.get_hdd_details(snapshot, device)
-
-        data_live = {'usage_time_hdd': usage_time_hdd,
-                     'serial_number': serial_number,
-                     'snapshot_uuid': snapshot['uuid'],
-                     'description': '',
-                     'device': device}
-
-        live = Live(**data_live)
-
-        if not usage_time_hdd:
-            warning = f"We don't found any TestDataStorage for disk sn: {serial_number}"
-            live.severity = Severity.Warning
-            live.description = warning
-            return live
-
-        live.sort_actions()
-        diff_time = live.diff_time()
-        if diff_time is None:
-            warning = "Don't exist one previous live or snapshot as reference"
-            live.description += warning
-            live.severity = Severity.Warning
-        elif diff_time < timedelta(0):
-            warning = "The difference with the last live/snapshot is negative"
-            live.description += warning
-            live.severity = Severity.Warning
-        return live
-
     def transfer_ownership(self):
         """Perform a InitTransfer action to change author_id of device"""
         pass

tests/test_action.py

@@ -262,11 +262,12 @@ def test_live(user: UserClient, app: Devicehub):
     }
 
     user.post(res=models.Allocate, data=post_request)
-    acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
     hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
     hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
     hdd_action['lifetime'] += 1000
-    snapshot, _ = user.post(acer, res=models.Snapshot)
+    acer.pop('elapsed')
+    acer['licence_version'] = '1.0.0'
+    snapshot, _ = user.post(acer, res=models.Live)
     db_device = Device.query.filter_by(id=1).one()
     action_live = [a for a in db_device.actions if a.type == 'Live']
     assert len(action_live) == 1
@@ -274,6 +275,7 @@ def test_live(user: UserClient, app: Devicehub):
     assert action_live[0].usage_time_allocate == timedelta(hours=1000)
     assert action_live[0].final_user_code == post_request['finalUserCode']
     assert action_live[0].serial_number == 'wd-wx11a80w7430'
+    assert action_live[0].licence_version == '1.0'
     assert str(action_live[0].snapshot_uuid) == acer['uuid']
 
 @pytest.mark.mvp
@@ -297,9 +299,9 @@ def test_live_without_TestDataStorage(user: UserClient, app: Devicehub):
     acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
     actions = [a for a in acer['components'][7]['actions'] if a['type'] != 'TestDataStorage']
     acer['components'][7]['actions'] = actions
-    live, _ = user.post(acer, res=models.Snapshot)
+    acer.pop('elapsed')
-    assert live['type'] == 'Live'
+    acer['licence_version'] = '1.0.0'
-    assert live['serialNumber'] == 'wd-wx11a80w7430'
+    live, _ = user.post(acer, res=models.Live)
     assert live['severity'] == 'Warning'
     description = "We don't found any TestDataStorage for disk sn: wd-wx11a80w7430"
     assert live['description'] == description
@@ -328,7 +330,9 @@ def test_live_without_hdd_1(user: UserClient, app: Devicehub):
     acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
     components = [a for a in acer['components'] if a['type'] != 'HardDrive']
     acer['components'] = components
-    response, _ = user.post(acer, res=models.Snapshot, status=404)
+    acer.pop('elapsed')
+    acer['licence_version'] = '1.0.0'
+    response, _ = user.post(acer, res=models.Live, status=404)
     assert "The There aren't any disk in this device" in response['message']
 
 
@@ -353,7 +357,9 @@ def test_live_without_hdd_2(user: UserClient, app: Devicehub):
 
     user.post(res=models.Allocate, data=post_request)
     acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
-    response, _ = user.post(acer, res=models.Snapshot, status=404)
+    acer.pop('elapsed')
+    acer['licence_version'] = '1.0.0'
+    response, _ = user.post(acer, res=models.Live, status=404)
     assert "The There aren't any disk in this device" in response['message']
 
 
@@ -380,9 +386,9 @@ def test_live_without_hdd_3(user: UserClient, app: Devicehub):
 
     user.post(res=models.Allocate, data=post_request)
     acer = file('acer.happy.battery.snapshot')
-    live, _ = user.post(acer, res=models.Snapshot)
+    acer.pop('elapsed')
-    assert live['type'] == 'Live'
+    acer['licence_version'] = '1.0.0'
-    assert live['serialNumber'] == 'wd-wx11a80w7430'
+    live, _ = user.post(acer, res=models.Live)
     assert live['severity'] == 'Warning'
     description = "Don't exist one previous live or snapshot as reference"
     assert live['description'] == description
@@ -414,9 +420,9 @@ def test_live_with_hdd_with_old_time(user: UserClient, app: Devicehub):
     acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
     action = [a for a in acer['components'][7]['actions'] if a['type'] == 'TestDataStorage']
     action[0]['lifetime'] -= 100
-    live, _ = user.post(acer, res=models.Snapshot)
+    acer.pop('elapsed')
-    assert live['type'] == 'Live'
+    acer['licence_version'] = '1.0.0'
-    assert live['serialNumber'] == 'wd-wx11a80w7430'
+    live, _ = user.post(acer, res=models.Live)
     assert live['severity'] == 'Warning'
     description = "The difference with the last live/snapshot is negative"
     assert live['description'] == description
@@ -446,11 +452,14 @@ def test_live_search_last_allocate(user: UserClient, app: Devicehub):
     hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
     hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
     hdd_action['lifetime'] += 1000
-    live, _ = user.post(acer, res=models.Snapshot)
+    acer.pop('elapsed')
+    acer['licence_version'] = '1.0.0'
+    live, _ = user.post(acer, res=models.Live)
     acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec4"
     actions = [a for a in acer['components'][7]['actions'] if a['type'] != 'TestDataStorage']
     acer['components'][7]['actions'] = actions
-    live, _ = user.post(acer, res=models.Snapshot)
+    live, _ = user.post(acer, res=models.Live)
+    import pdb; pdb.set_trace()
     assert live['usageTimeAllocate'] == 1000