fox some tasks

Author: maybe-hello-world

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "netunicorn-library"
-version = "0.3.0"
+version = "0.3.1"
 authors = [
     {name = "Roman Beltiukov", email = "[email protected]"},
 ]

tasks/measurements/ping.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass
 from typing import List
 
+from netunicorn.base.architecture import Architecture
 from netunicorn.base.nodes import Node
 from netunicorn.base.task import Failure, Task, TaskDispatcher
 
@@ -35,10 +36,10 @@ def __init__(self, address: str, count: int = 1, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def dispatch(self, node: Node) -> Task:
-        if node.properties.get("os_family", "").lower() == "linux":
+        if node.architecture in {Architecture.LINUX_AMD64, Architecture.LINUX_ARM64}:
             return PingLinuxImplementation(self.address, self.count)
         raise NotImplementedError(
-            f'Ping is not implemented for {node.properties.get("os_family", "")}'
+            f'Ping is not implemented for architecture: {node.architecture}'
         )
 
 

tasks/qoe_youtube/__init__.py

@@ -81,7 +81,7 @@ def __init__(self, start_task_name: str, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def dispatch(self, node: Node) -> Task:
-        if node.properties.get("os_family", "").lower() == "linux":
+        if node.architecture in {Architecture.LINUX_AMD64, Architecture.LINUX_ARM64}:
             return StopQoECollectionServerLinuxImplementation(start_task_name=self.start_task_name, name=self.name)
 
         raise NotImplementedError(
@@ -137,7 +137,7 @@ def dispatch(self, node: Node) -> Task:
             )
 
         raise NotImplementedError(
-            f'WatchYouTubeVideo is not implemented for {node.properties.get("os_family", "")}'
+            f'WatchYouTubeVideo is not implemented for architecture: {node.architecture}'
         )
 
 

tasks/video_watchers/twitch_watcher.py

@@ -8,6 +8,7 @@
 from typing import Optional
 
 from netunicorn.base import Result, Success, Task, TaskDispatcher
+from netunicorn.base.architecture import Architecture
 from netunicorn.base.nodes import Node
 
 from selenium import webdriver
@@ -48,11 +49,11 @@ def __init__(self, video_url: str, duration: Optional[int] = None, *args, **kwar
         super().__init__(*args, **kwargs)
 
     def dispatch(self, node: Node) -> Task:
-        if node.properties.get("os_family", "").lower() == "linux":
+        if node.architecture in {Architecture.LINUX_AMD64, Architecture.LINUX_ARM64}:
             return WatchTwitchStreamLinuxImplementation(self.video_url, self.duration, name=self.name)
 
         raise NotImplementedError(
-            f'WatchYouTubeVideo is not implemented for {node.properties.get("os_family", "")}'
+            f'WatchTwitchVideo is not implemented for architecture: {node.architecture}'
         )
 
 

tasks/video_watchers/vimeo_watcher.py

@@ -8,6 +8,7 @@
 from typing import Optional
 
 from netunicorn.base import Result, Failure, Success, Task, TaskDispatcher
+from netunicorn.base.architecture import Architecture
 from netunicorn.base.nodes import Node
 
 from selenium import webdriver
@@ -78,11 +79,11 @@ def __init__(self, video_url: str, duration: Optional[int] = None, *args, **kwar
         super().__init__(*args, **kwargs)
 
     def dispatch(self, node: Node) -> Task:
-        if node.properties.get("os_family", "").lower() == "linux":
+        if node.architecture in {Architecture.LINUX_AMD64, Architecture.LINUX_ARM64}:
             return WatchVimeoVideoLinuxImplementation(self.video_url, self.duration, name=self.name)
 
         raise NotImplementedError(
-            f'WatchYouTubeVideo is not implemented for {node.properties.get("os_family", "")}'
+            f'WatchVimeoVideo is not implemented for architecture: {node.architecture}'
         )
 
 

tasks/video_watchers/youtube_watcher.py

@@ -9,6 +9,7 @@
 from enum import IntEnum
 
 from netunicorn.base import Result, Failure, Success, Task, TaskDispatcher
+from netunicorn.base.architecture import Architecture
 from netunicorn.base.nodes import Node
 
 from selenium import webdriver
@@ -101,11 +102,11 @@ def __init__(self, video_url: str, duration: Optional[int] = None, *args, **kwar
         super().__init__(*args, **kwargs)
 
     def dispatch(self, node: Node) -> Task:
-        if node.properties.get("os_family", "").lower() == "linux":
+        if node.architecture in {Architecture.LINUX_AMD64, Architecture.LINUX_ARM64}:
             return WatchYouTubeVideoLinuxImplementation(self.video_url, self.duration, name=self.name)
 
         raise NotImplementedError(
-            f'WatchYouTubeVideo is not implemented for {node.properties.get("os_family", "")}'
+            f'WatchYouTubeVideo is not implemented for architecture: {node.architecture}'
         )