Merge pull request #85 from eReuse/feature/84-new-hid

Feature/84 new hid

CHANGELOG.md

@@ -11,6 +11,9 @@ ml).
 ## testing
   [1.0.2-beta]
 
+## [1.0.3-beta]
+- [addend] #85 add mac of network adapter to device hid
+
 ## [1.0.2-beta]
 - [addend] #87 allocate, deallocate and live actions
 - [fixed] #89 save json on disk only for shapshots

ereuse_devicehub/__init__.py

@@ -1 +1 @@
-__version__ = "1.0.2-beta"
+__version__ = "1.0.3-beta"

ereuse_devicehub/migrations/versions/bf600ca861a4_adding_hid.py

@@ -0,0 +1,65 @@
+"""empty message
+
+Revision ID: bf600ca861a4
+Revises: 68a5c025ab8e
+Create Date: 2020-12-15 15:58:41.545563
+
+"""
+from alembic import context
+from alembic import op
+import sqlalchemy as sa
+import sqlalchemy_utils
+import citext
+import teal
+
+
+# revision identifiers, used by Alembic.
+revision = 'bf600ca861a4'
+down_revision = '68a5c025ab8e'
+branch_labels = None
+depends_on = None
+
+
+def get_inv():
+    INV = context.get_x_argument(as_dictionary=True).get('inventory')
+    if not INV:
+        raise ValueError("Inventory value is not specified")
+    return INV
+
+def upgrade():
+    con = op.get_bind()
+    sql = f"""
+    select d.id, d.hid, dd.serial_number from {get_inv()}.computer as c 
+        join {get_inv()}.device as d on c.id=d.id 
+        inner join {get_inv()}.component as cmp on cmp.parent_id=c.id 
+        inner join {get_inv()}.network_adapter as net on net.id=cmp.id 
+        join {get_inv()}.device as dd on net.id=dd.id;
+    """
+    computers = con.execute(sql)
+    hids = {}
+    macs = {}
+    for c in computers:
+        hids[c.id] = c.hid
+        if not c.serial_number:
+            continue
+        try:
+            macs[c.id].append(c.serial_number)
+            macs[c.id].sort()
+        except:
+            macs[c.id] = [c.serial_number]
+
+    for id_dev, hid in hids.items():
+        if not (id_dev and hid):
+            continue
+        if not id_dev in macs:
+            continue
+        mac = macs[id_dev][0]
+        new_hid = "{}-{}".format(hid, mac)
+
+        sql = f"update {get_inv()}.device set hid='{new_hid}' where id={id_dev};"
+        con.execute(sql)
+
+
+
+def downgrade():
+    pass

ereuse_devicehub/resources/action/models.py

@@ -1547,6 +1547,8 @@ def update_components_action_one(target: ActionWithOneDevice, device: Device, __
         target.components.clear()
         if isinstance(device, Computer):
             target.components |= device.components
+    elif isinstance(device, Computer):
+            device.add_mac_to_hid()
 
 
 @event.listens_for(ActionWithMultipleDevices.devices, Events.init_collection.__name__,

ereuse_devicehub/resources/action/views.py

@@ -15,6 +15,9 @@ from teal.resource import View
 from teal.db import ResourceNotFound
 
 from ereuse_devicehub.db import db
+from ereuse_devicehub.resources.device.models import Device, Computer
+from ereuse_devicehub.resources.action.models import Action, RateComputer, Snapshot, VisualTest, \
+    InitTransfer
 from ereuse_devicehub.query import things_response
 from ereuse_devicehub.resources.action.models import (Action, RateComputer, Snapshot, VisualTest, 
     InitTransfer, Live, Allocate, Deallocate)
@@ -157,6 +160,8 @@ class ActionView(View):
         components = None
         if snapshot_json['software'] == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid):
             components = snapshot_json.pop('components', None)  # type: List[Component]
+            if isinstance(device, Computer) and device.hid:
+                device.add_mac_to_hid(components_snap=components)
         snapshot = Snapshot(**snapshot_json)
 
         # Remove new actions from devices so they don't interfere with sync
@@ -234,17 +239,34 @@ class ActionView(View):
             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."""
-        device = snapshot.get('device')  # type: Computer
+        hid = self.get_hid(snapshot)
-        # TODO @cayop dependency of pulls 85
+        if not hid or not Device.query.filter(
-        # if the pr/85 is merged, then you need change this way for get the device
+            Device.hid==hid, Device.owner_id==g.user.id).count():
-        if not device.hid or not Device.query.filter(
-            Device.hid==device.hid, Device.owner_id==g.user.id).count():
             return None
 
         device = Device.query.filter(
-            Device.hid==device.hid, Device.owner_id==g.user.id).one()
+            Device.hid==hid, Device.owner_id==g.user.id).one()
 
         if not device.allocated:
             return None

ereuse_devicehub/resources/device/models.py

@@ -146,8 +146,7 @@ class Device(Thing):
 
     def __init__(self, **kw) -> None:
         super().__init__(**kw)
-        with suppress(TypeError):
+        self.set_hid()
-            self.hid = Naming.hid(self.type, self.manufacturer, self.model, self.serial_number)
 
     @property
     def actions(self) -> list:
@@ -298,6 +297,10 @@ class Device(Thing):
             args[POLYMORPHIC_ON] = cls.type
         return args
 
+    def set_hid(self):
+        with suppress(TypeError):
+            self.hid = Naming.hid(self.type, self.manufacturer, self.model, self.serial_number)
+
     def last_action_of(self, *types):
         """Gets the last action of the given types.
 
@@ -483,6 +486,23 @@ class Computer(Device):
             if privacy
         )
 
+    def add_mac_to_hid(self, components_snap=None):
+        """Returns the Naming.hid with the first mac of network adapter, 
+        following an alphabetical order.
+        """
+        self.set_hid()
+        if not self.hid:
+            return
+        components = self.components if components_snap is None else components_snap
+        macs_network = [c.serial_number for c in components
+                        if c.type == 'NetworkAdapter' and c.serial_number is not None]
+        macs_network.sort()
+        mac = macs_network[0] if macs_network else ''
+        if not mac or mac in self.hid:
+            return
+        mac = f"-{mac}"
+        self.hid += mac
+
     def __format__(self, format_spec):
         if not format_spec:
             return super().__format__(format_spec)

ereuse_devicehub/resources/device/views.py

@@ -241,6 +241,7 @@ class DeviceMergeView(View):
                 setattr(self.base_device, field_name, value)
 
         self.base_device.hid = self.with_device.hid
+        self.base_device.add_mac_to_hid()
 
 
 class ManufacturerView(View):

tests/files/4-first-device-but-removing-processor.snapshot-and-adding-graphic-card.yaml

@@ -5,11 +5,6 @@ device:
   type: Desktop
   chassis: Tower
 components:
-  - manufacturer: p1c4m
-    serialNumber: p1c4s
-    type: NetworkAdapter
-    speed: 1000
-    wireless: False
   - manufacturer: p1c3m
     serialNumber: p1c3s
     type: GraphicCard

tests/test_device.py

@@ -1,8 +1,10 @@
+import copy
 import datetime
+import pytest
+
 from uuid import UUID
 from flask import g
 
-import pytest
 from colour import Color
 from ereuse_utils.naming import Naming
 from ereuse_utils.test import ANY
@@ -608,3 +610,111 @@ def test_cooking_mixer_api(user: UserClient):
     mixer, _ = user.get(res=d.Device, item=snapshot['device']['id'])
     assert mixer['type'] == 'Mixer'
     assert mixer['serialNumber'] == 'foo'
+
+
+@pytest.mark.mvp
+def test_hid_with_mac(app: Devicehub, user: UserClient):
+    """Checks hid with mac."""
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    user.post(snapshot, res=m.Snapshot)
+    pc, _ = user.get(res=d.Device, item=1)
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
+
+
+@pytest.mark.mvp
+def test_hid_without_mac(app: Devicehub, user: UserClient):
+    """Checks hid without mac."""
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    snapshot['components'] = [c for c in snapshot['components'] if c['type'] != 'NetworkAdapter']
+    user.post(snapshot, res=m.Snapshot)
+    pc, _ = user.get(res=d.Device, item=1)
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
+
+
+@pytest.mark.mvp
+def test_hid_with_mac_none(app: Devicehub, user: UserClient):
+    """Checks hid with mac = None."""
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0]
+    network['serialNumber'] = None
+    user.post(snapshot, res=m.Snapshot)
+    pc, _ = user.get(res=d.Device, item=1)
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
+
+
+@pytest.mark.mvp
+def test_hid_with_2networkadapters(app: Devicehub, user: UserClient):
+    """Checks hid with 2 networks adapters"""
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0]
+    network2 = copy.copy(network)
+    snapshot['components'].append(network2)
+    network['serialNumber'] = 'a0:24:8c:7f:cf:2d'
+    user.post(snapshot, res=m.Snapshot)
+    devices, _ = user.get(res=d.Device)
+
+    laptop = devices['items'][0]
+    assert laptop['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
+    assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 1
+
+
+@pytest.mark.mvp
+def test_hid_with_2network_and_drop_no_mac_in_hid(app: Devicehub, user: UserClient):
+    """Checks hid with 2 networks adapters and next drop the network is not used in hid"""
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0]
+    network2 = copy.copy(network)
+    snapshot['components'].append(network2)
+    network['serialNumber'] = 'a0:24:8c:7f:cf:2d'
+    user.post(snapshot, res=m.Snapshot)
+    pc, _ = user.get(res=d.Device, item=1)
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
+
+    snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb'
+    snapshot['components'] = [c for c in snapshot['components'] if c != network]
+    user.post(snapshot, res=m.Snapshot)
+    devices, _ = user.get(res=d.Device)
+    laptop = devices['items'][0]
+    assert laptop['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
+    assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 1
+    assert len([c for c in laptop['components'] if c['type'] == 'NetworkAdapter']) == 1
+
+
+@pytest.mark.mvp
+def test_hid_with_2network_and_drop_mac_in_hid(app: Devicehub, user: UserClient):
+    """Checks hid with 2 networks adapters and next drop the network is used in hid"""
+    # One tipical snapshot with 2 network cards
+    snapshot = file('asus-eee-1000h.snapshot.11')
+    network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0]
+    network2 = copy.copy(network)
+    snapshot['components'].append(network2)
+    network['serialNumber'] = 'a0:24:8c:7f:cf:2d'
+    user.post(snapshot, res=m.Snapshot)
+    pc, _ = user.get(res=d.Device, item=1)
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
+
+    # we drop the network card then is used for to build the hid
+    snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb'
+    snapshot['components'] = [c for c in snapshot['components'] if c != network2]
+    user.post(snapshot, res=m.Snapshot)
+    devices, _ = user.get(res=d.Device)
+    laptops = [c for c in devices['items'] if c['type'] == 'Laptop']
+    assert len(laptops) == 2
+    hids = [h['hid'] for h in laptops]
+    proof_hid = ['laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d',
+                 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d']
+    assert all([h in proof_hid for h in hids])
+
+    # we drop all network cards
+    snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abc'
+    snapshot['components'] = [c for c in snapshot['components'] if not c in [network, network2]]
+    user.post(snapshot, res=m.Snapshot)
+    devices, _ = user.get(res=d.Device)
+    laptops = [c for c in devices['items'] if c['type'] == 'Laptop']
+    assert len(laptops) == 3
+    hids = [h['hid'] for h in laptops]
+    proof_hid = ['laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d',
+                 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d',
+                 'laptop-asustek_computer_inc-1000h-94oaaq021116']
+    assert all([h in proof_hid for h in hids])
+

tests/test_snapshot.py

@@ -260,7 +260,7 @@ def test_snapshot_component_add_remove(user: UserClient):
     # We register the first device but without the processor,
     # adding a graphic card and adding a new component
     s4 = file('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card')
-    snapshot_and_check(user, s4, ('RateComputer',), perform_second_snapshot=False)
+    snapshot4 = snapshot_and_check(user, s4, ('RateComputer',), perform_second_snapshot=False)
     pc1, _ = user.get(res=m.Device, item=pc1_id)
     pc2, _ = user.get(res=m.Device, item=pc2_id)
     # Check if the update_timestamp is updated
@@ -269,12 +269,10 @@ def test_snapshot_component_add_remove(user: UserClient):
     assert not update4_pc1 in [update1_pc1, update2_pc1, update3_pc1]
     assert update3_pc2 == update2_pc2
     # PC 0: p1c3s, p1c4s. PC1: p2c1s
-    assert {c['serialNumber'] for c in pc1['components']} == {'p1c3s', 'p1c4s'}
+    assert {c['serialNumber'] for c in pc1['components']} == {'p1c3s'}
     assert all(c['parent'] == pc1_id for c in pc1['components'])
     # This last Action only
-    act = get_actions_info(pc1['actions'])[-1]
+    assert get_actions_info(pc1['actions'])[-1] == ('RateComputer', ['p1c3s'])
-    assert 'RateComputer' in act
-    assert set(act[1]) == {'p1c3s', 'p1c4s'}
     # PC2
     # We haven't changed PC2
     assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)

tests/test_workbench.py

@@ -52,7 +52,7 @@ def test_workbench_server_condensed(user: UserClient):
     device, _ = user.get(res=Device, item=snapshot['device']['id'])
     assert device['dataStorageSize'] == 1100
     assert device['chassis'] == 'Tower'
-    assert device['hid'] == 'desktop-d1mr-d1ml-d1s'
+    assert device['hid'] == 'desktop-d1mr-d1ml-d1s-na1-s'
     assert device['graphicCardModel'] == device['components'][0]['model'] == 'gc1-1ml'
     assert device['networkSpeeds'] == [1000, 58]
     assert device['processorModel'] == device['components'][3]['model'] == 'p1-1ml'
@@ -138,7 +138,7 @@ def test_real_hp_11(user: UserClient):
     s = file('real-hp.snapshot.11')
     snapshot, _ = user.post(res=em.Snapshot, data=s)
     pc = snapshot['device']
-    assert pc['hid'] == 'desktop-hewlett-packard-hp_compaq_8100_elite_sff-czc0408yjg'
+    assert pc['hid'] == 'desktop-hewlett-packard-hp_compaq_8100_elite_sff-czc0408yjg-6c:62:6d:81:22:9f'
     assert pc['chassis'] == 'Tower'
     assert set(e['type'] for e in snapshot['actions']) == {
         'EreusePrice',
@@ -179,7 +179,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
     assert pc['model'] == '1001pxd'
     assert pc['serialNumber'] == 'b8oaas048286'
     assert pc['manufacturer'] == 'asustek computer inc.'
-    assert pc['hid'] == 'laptop-asustek_computer_inc-1001pxd-b8oaas048286'
+    assert pc['hid'] == 'laptop-asustek_computer_inc-1001pxd-b8oaas048286-14:da:e9:42:f6:7c'
     assert pc['tags'] == []
     assert pc['networkSpeeds'] == [100, 0], 'Although it has WiFi we do not know the speed'
     assert pc['rate']