Merge branch 'release-24.8' into 'main'

Release 24.9

See merge request nwac/sdk-py!130

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## Version 4.0.0
+
+Breaking changes:
+- `device.create_qod_session()` now requires `duration` as a mandatory parameter
+- `device.get_congestion()` now returns a list of `Congestion` objects
+
+Changes:
+- `device.verify_location()` may now return a "PARTIAL" result if the device is
+  partially inside the verification area
+
+Fixes:
+- Previously due to a miscommunication `device.sessions()` would return all
+  created sessions. These have now been correctly limited to device-specific ones
+
 ## Version 3.1.0
 
 Changes:

Jenkinsfile

@@ -93,6 +93,17 @@ pipeline {
                 }        
             }
         }
+        stage('Audit') {
+            steps {
+                container('beluga') {
+                    script {
+                        sh """
+                            https_proxy="http://fihel1d-proxy.emea.nsn-net.net:8080" python3 -m poetry run pip-audit
+                        """
+                    }
+                }
+            }
+        }
         stage('Integration Test') {
             steps {
                 container('beluga') {
@@ -125,17 +136,6 @@ pipeline {
                 }
             }
         }
-        stage('Audit') {
-            steps {
-                container('beluga') {
-                    script {
-                        sh """
-                            https_proxy="http://fihel1d-proxy.emea.nsn-net.net:8080" python3 -m poetry run pip-audit
-                        """
-                    }
-                }
-            }
-        }
         stage('Build') {
             steps {
                 container('beluga') {
@@ -150,7 +150,7 @@ pipeline {
         }
         stage('Installation Test') {
             when { expression { env.gitlabActionType == "TAG_PUSH" && 
-            (env.gitlabTargetBranch.contains("rc-") || env.gitlabTargetBranch.contains("release-"))} }
+            (env.gitlabBranch.contains("rc-") || env.gitlabBranch.contains("release-"))} }
             steps {
                 container('beluga') {
                     script {
@@ -167,14 +167,14 @@ pipeline {
             }
         }
         stage('Deploy candidate') {
-            when { expression { env.gitlabActionType == "TAG_PUSH" && env.gitlabTargetBranch.contains("rc-")} }
+            when { expression { env.gitlabActionType == "TAG_PUSH" && env.gitlabBranch.contains("rc-")} }
             steps {
                 container('beluga') {
                     script {
                         sh """
                         env | grep gitlab
                         """
-                        if(env.gitlabActionType == "TAG_PUSH" && env.gitlabTargetBranch.contains("rc-")){
+                        if(env.gitlabActionType == "TAG_PUSH" && env.gitlabBranch.contains("rc-")){
                             sh '''
                                 python3 -m poetry config repositories.devpi ${PYPI_REPOSITORY}
                                 python3 -m poetry build
@@ -186,7 +186,7 @@ pipeline {
             }
         }
         stage('Deploy release') {
-            when { expression { env.gitlabActionType == "TAG_PUSH" && env.gitlabTargetBranch.contains("release-")} }
+            when { expression { env.gitlabActionType == "TAG_PUSH" && env.gitlabBranch.contains("release-")} }
             steps {
                 container('beluga') {
                     script {

examples/device-status.py

@@ -1,39 +1,47 @@
-# Device Status functionalities
-
-# Subscribing to Connectivity and Roaming updates:
-
-from typing import List
 import network_as_code as nac
+from datetime import datetime, timedelta, timezone
+from network_as_code.models.device import DeviceIpv4Addr
 
-from network_as_code.models.device import Device, DeviceIpv4Addr
-
+# We begin by creating a Network as Code client
 client = nac.NetworkAsCodeClient(
     token="<your-application-key-here>"
 )
 
-device = client.devices.get(
-    "[email protected]",
-    ipv4_address = DeviceIpv4Addr(
-        public_address="233.252.0.2",
-        private_address="192.0.2.25",
-        public_port=80),
-    ipv6_address = "2001:db8:1234:5678:9abc:def0:fedc:ba98",
+# Then, we create an object for the mobile device we want to use
+my_device = client.devices.get(
+    "[email protected]",
+    ipv4_address=DeviceIpv4Addr(
+        public_address="192.0.2.3",
+        private_address="192.0.2.204",
+        public_port=80
+    ),
+    ipv6_address="2001:db8:1234:5678:9abc:def0:fedc:ba98",
     # The phone number accepts the "+" sign, but not spaces or "()" marks
-    phone_number = "36721601234567"
+    phone_number="36721601234567"
 )
 
-connectivity_subscription = client.connectivity.subscribe(
-    # Change it to "ROAMING_STATUS" whenever needed
-    event_type="CONNECTIVITY",
-    device=device,
-    max_num_of_reports=5,
-    notification_url="https://example.com/notifications",
+# Simply change the event_type to
+# "org.camaraproject.device-status.v0.roaming-status" whenever needed.
+my_subscription = client.connectivity.subscribe(
+    event_type="org.camaraproject.device-status.v0.connectivity-data",
+    device=my_device,
+    # You can tell when the subscription is supposed to expire
+    # with a date-time object
+    subscription_expire_time=datetime.now(timezone.utc) + timedelta(days=1),
     # Use HTTPS to send notifications
+    notification_url="https://example.com/notifications",
     notification_auth_token="replace-with-your-auth-token"
 )
 
-# Get a subscription by its ID
-subscription = client.connectivity.get_subscription(connectivity_subscription.id)
+# Use this to show the roaming subscription status
+print(my_subscription)
+
+# You can also get a subscription by its ID
+subscription = client.connectivity.get_subscription(my_subscription.id)
+
+# Or check when your subscription starts/expires:
+print(subscription.starts_at)
+print(subscription.expires_at)
 
 # Delete a subscription
-connectivity_subscription.delete()
+subscription.delete()

examples/identity-security.py

@@ -14,13 +14,22 @@
 # Then, create a device object for the phone number you want to check
 my_device = client.devices.get(
     # The phone number accepts the "+" sign, but not spaces or "()" marks
-    phone_number="36721601234567"
+    phone_number="+346661113334"
 )
 
-# Check the latest SIM-Swap date
-latest_sim_swap_date = my_device.get_sim_swap_date()
+# The date of the last SIM Swap can be retrieved like so:
+# The output may be null, if no SIM Swap has occurred.
+# Or it may also return the SIM activation date.
+sim_swap_date = my_device.get_sim_swap_date()
 
- # Check SIM-Swap events within specified time spans
-# The max_age parameter is not mandatory
-# This method also checks if SIM swap occurred within an undefined age
-sim_swap_check = my_device.verify_sim_swap(max_age=360)
+# Otherwise it behaves like a regular datetime object
+print(sim_swap_date.isoformat())
+
+# If you are only interested if a SIM swap has occurred,
+# just use:
+if my_device.verify_sim_swap():
+    print("There has been a SIM swap!")
+
+# You can also test if the SIM swap happened recently:
+if my_device.verify_sim_swap(max_age=3600):
+    print("A SIM swap occurred within the past hour!")

examples/location-information.py

@@ -16,7 +16,7 @@
 
 from network_as_code.models.device import Device, DeviceIpv4Addr
 
-SDK_TOKEN = "<replace-me>"
+SDK_TOKEN = "<your-application-key-here>"
 DEVICE_ID = "[email protected]"
 
 # Give the device the device identifier and SDK token

examples/network-insights.py

@@ -9,7 +9,7 @@
 from network_as_code.models.device import Device, DeviceIpv4Addr
 
 # Initialize the client object with your application key
-SDK_TOKEN = "<replace-me>"
+SDK_TOKEN = "<your-application-key-here>"
 DEVICE_ID = "[email protected]"
 
 # Give the device the device identifier and SDK token
@@ -19,7 +19,8 @@
 
 my_device = client.devices.get(DEVICE_ID)
 
-# Subscribe your device to Congestion notifications
+# Subscribe your device to Congestion notifications FIRST.
+# Then, you'll be able to use other functionalities, such as polling.
 congestion_subscription = client.insights.subscribe_to_congestion_info(
     my_device,
     # Set the duration of your subscription to congestion insights,
@@ -38,9 +39,13 @@
 print(congestion_subscription.starts_at)
 print(congestion_subscription.expires_at)
 
-# Get historical data between two timestamps
+# Get congestion historical data between two timestamps
 # Set the duration/time difference with the timedelta function
+# It returns an array of congested objects and the prediction confidence level
 congestion = my_device.get_congestion(
     start=datetime.now(timezone.utc),
     end=datetime.now(timezone.utc) + timedelta(hours=3)
 )
+
+# Show the congestion level objects
+print(congestion)

examples/network_slices.py

@@ -10,12 +10,11 @@
     Point,
     AreaOfService,
     NetworkIdentifier,
-    Slice,
     SliceInfo,
     Throughput
 )
 
-SDK_TOKEN = "<replace-me>"
+SDK_TOKEN = "<your-application-key-here>"
 DEVICE_ID = "[email protected]"
 
 
@@ -24,31 +23,37 @@
     token=SDK_TOKEN
 )
 
-
 # Creation of a slice
 # We use the country code (MCC) and network code (MNC) to identify the network
 # Different types of slices can be requested using service type and differentiator
 # Area of the slice must also be described in geo-coordinates
 my_slice = client.slices.create(
-    name="slice-name",
-    network_id=NetworkIdentifier(mcc="664", mnc="22"),
-    slice_info=SliceInfo(service_type="eMBB", differentiator="123456"),
-    area_of_service=AreaOfService(
-        polygon=[
-            Point(latitude=42.0, longitude=42.0),
-            Point(latitude=41.0, longitude=42.0),
-            Point(latitude=42.0, longitude=41.0),
-            Point(latitude=42.0, longitude=42.0)
-        ]
-    ),
-    notification_url="https://notify.me/here",
-    # Use HTTPS to send notifications
-    notification_auth_token="replace-with-your-auth-token"
+    name="slice-test-2",
+    network_id = NetworkIdentifier(mcc="236", mnc="30"),
+    slice_info = SliceInfo(service_type="eMBB", differentiator="444444"),
+    area_of_service = AreaOfService(polygon=[
+        Point(latitude=47.344, longitude=104.349),
+        Point(latitude=35.344, longitude=76.619),
+        Point(latitude=12.344, longitude=142.541),
+        Point(latitude=19.43, longitude=103.53)
+    ]),
+    slice_downlink_throughput = Throughput(guaranteed=3415, maximum=1234324),
+    slice_uplink_throughput = Throughput(guaranteed=3415, maximum=1234324),
+    device_downlink_throughput = Throughput(guaranteed=3415, maximum=1234324),
+    device_uplink_throughput = Throughput(guaranteed=3415, maximum=1234324),
+    max_data_connections=10,
+    max_devices=6,
+     # Use HTTPS to send notifications
+    notification_url="https://snippets.requestcatcher.com/test",
 )
 
 # Get a slice by its ID
 slice = client.slices.get(my_slice.name)
 
+# Get a slice by using an index
+# or remove the index '[0]' to get all slices
+one_slice = client.slices.get_all()[0]
+
 # Modify the slice
 my_slice.modify(
     max_data_connections = 12,
@@ -89,4 +94,4 @@ async def slice_attachments():
 # Simple manual polling can be implemented like this:
 while my_slice.state != "AVAILABLE":
     my_slice.refresh()
-    time.sleep(1)
+    time.sleep(1)

examples/qod_example.py

@@ -6,7 +6,7 @@
 
 from network_as_code.models.device import DeviceIpv4Addr
 
-client = nac.NetworkAsCodeClient(...)
+client = nac.NetworkAsCodeClient("<your-application-key-here>")
 
 # Identify the device with its ID,
 # IP address(es) and optionally, a phone number

integration_tests/test_device_status.py

@@ -21,8 +21,6 @@ def test_creating_connectivity_subscription_with_notification(client, device):
         notification_auth_token="c8974e592c2fa383d4a3960714",
     )
 
-    print(subscription)
-
     subscription.delete()
 
 def test_creating_connectivity_subscription_roaming(client, device):