Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,10 @@ def __init__(self, settings = None, usage_sharing = True, **kwargs):
@type restricted_properties: list
@param restricted_properties: List of property keys to limit the engine to
@type license_keys: string
@param licence_keys: licencekeys to use for the data file update service
@param licence_keys: licence keys to use for the data file update service.
Needed only when auto_update or update_on_start is on with the default
data_update_url, and the engine refuses to build without one then. The
file system watcher needs no key.
@type download: bool
@param download: whether to download the datafile or keep it in memory when it is returned from the datafile update service
@type max_matched_useragent_length: int
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,16 @@
from .swig_data import SwigData
from .devicedetection_datafile import DeviceDetectionDataFile

# The Distributor, which serves data files only to a licence key.
DISTRIBUTOR_URL = "https://distributor.51degrees.com/api/v2/download"

LICENCE_KEY_REQUIRED_FOR_UPDATES = (
"Some enabled features ({}) require a licence key but one has not been "
"supplied. Supply one with licence_keys, or set {} to False to build "
"without one. The file system watcher needs no licence key. A key can be "
"obtained from the 51Degrees website: "
"https://51degrees.com/pricing?utm_source=code&utm_medium=comment&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_onpremise-src-fiftyone_devicedetection_onpremise-devicedetection_onpremise.py&utm_term=licence-key-required-for-updates")


class DeviceDetectionOnPremise(Engine):
"""!
Expand Down Expand Up @@ -66,7 +76,7 @@ def __init__(
reuse_temp_file=False,
concurrency=multiprocessing.cpu_count(),
update_on_start=False,
data_update_url="https://distributor.51degrees.com/api/v2/download",
data_update_url=DISTRIBUTOR_URL,
use_predictive_graph=None,
use_performance_graph=None,
data_update_use_url_formatter=True,
Expand All @@ -87,7 +97,10 @@ def __init__(
@type restricted_properties: list
@param restricted_properties: List of property keys to limit the engine to
@type license_keys: string
@param licence_keys: licencekeys to use for the data file update service
@param licence_keys: licence keys to use for the data file update service.
Needed only when auto_update or update_on_start is on with the default
data_update_url, and the engine refuses to build without one then. The
file system watcher needs no key.
@type download: bool
@param download: whether to download the datafile or keep it in memory when it is returned from the datafile update service
@type max_matched_useragent_length: int
Expand Down Expand Up @@ -151,6 +164,19 @@ def __init__(
if not data_file_path and not data:
raise Exception("data_file_path or data is required")

# Only polling and the update on start-up download from the
# Distributor, so only they need a licence key. The file system
# watcher needs none.
auto_update = bool(auto_update)
update_on_start = bool(update_on_start)
if data_update_url == DISTRIBUTOR_URL and not licence_keys:
need_key = [name for name, on in (
("auto_update", auto_update),
("update_on_start", update_on_start)) if on]
if need_key:
raise Exception(LICENCE_KEY_REQUIRED_FOR_UPDATES.format(
", ".join(need_key), " and ".join(need_key)))

if data_file_path:
if os.path.isfile(data_file_path) == False and not update_on_start:
raise Exception("There is no file at " + data_file_path)
Expand All @@ -162,9 +188,6 @@ def __init__(
else:
self.data = data

if not licence_keys and licence_keys != "":
raise Exception("licence key is required. A key can be obtained from the 51Degrees website: https://51degrees.com/pricing?utm_source=code&utm_medium=comment&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_onpremise-src-fiftyone_devicedetection_onpremise-devicedetection_onpremise.py&utm_term=licence-key-required. If you do not wish to use a key then you can specify an empty string, but this will cause automatic updates to be disabled.")

# Create SWIG wrapper vector for restricted properties and add

restricted_properties_vector = VectorStringSwig()
Expand Down Expand Up @@ -251,15 +274,6 @@ def __init__(
self.properties_list = properties_list
self.data_file_path = data_file_path

# Disable features that require a licence key if one was not supplied.

if licence_keys:
auto_update = auto_update and (len(licence_keys) > 0)
update_on_start = update_on_start and (len(licence_keys) > 0)
else:
auto_update = False
update_on_start = False

if auto_update or update_on_start or file_system_watcher:

# Construct DataFile
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,10 @@ def __init__(self, settings = None, usage_sharing = True, **kwargs):
@type restricted_properties: list
@param restricted_properties: List of property keys to limit the engine to
@type license_keys: string
@param licence_keys: licencekeys to use for the data file update service
@param licence_keys: licence keys to use for the data file update service.
Needed only when auto_update or update_on_start is on with the default
data_update_url, and the engine refuses to build without one then. The
file system watcher needs no key.
@type download: bool
@param download: whether to download the datafile or keep it in memory when it is returned from the datafile update service
@type max_matched_useragent_length: int
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
# *********************************************************************
# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
#
# This Original Work is licensed under the European Union Public Licence
# (EUPL) v.1.2 and is subject to its terms as set out below.
#
# If a copy of the EUPL was not distributed with this file, You can obtain
# one at https://opensource.org/licenses/EUPL-1.2.
#
# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
# amended by the European Commission) shall be deemed incompatible for
# the purposes of the Work and the provisions of the compatibility
# clause in Article 5 of the EUPL shall not apply.
#
# If using the Work as, or as part of, a network application, by
# including the attribution notice(s) required under Article 5 of the EUPL
# in the end user terms of the application under an appropriate heading,
# such notice(s) shall fulfill the requirements of that article.
# *********************************************************************

import unittest

from fiftyone_devicedetection_onpremise.devicedetection_onpremise import DeviceDetectionOnPremise
from fiftyone_devicedetection_onpremise.devicedetection_onpremise_pipelinebuilder import DeviceDetectionOnPremisePipelineBuilder

data_file = "../../cxx/data/device-detection/51Degrees-LiteV4.1.hash"


class LicenceKeyRuleTests(unittest.TestCase):
"""!
data-updates.md lines 107 to 109 and 392. A licence key is needed only
by polling (auto_update) and the update on start-up, which download from
the Distributor, and the build fails naming the option to turn off when
one of them is on with no key. The file system watcher needs no key.
"""

def build(self, **options):
engine = DeviceDetectionOnPremise(data_file_path=data_file, **options)
self.addCleanup(engine.close)
return engine

def test_no_key_is_needed_when_nothing_downloads(self):
for key in (None, ""):
with self.subTest(key=key):
engine = self.build(
licence_keys=key,
auto_update=False,
update_on_start=False,
file_system_watcher=True)
self.assertFalse(engine.data_file.auto_update)
self.assertFalse(engine.data_file.update_on_start)
self.assertTrue(engine.data_file.file_system_watcher)

def test_no_key_is_needed_with_the_engine_defaults(self):
self.build()

def test_a_download_with_no_key_is_refused_naming_the_option(self):
cases = [
(None, {"auto_update": True}, "(auto_update)"),
("", {"auto_update": True}, "(auto_update)"),
("", {"update_on_start": True}, "(update_on_start)"),
("", {"auto_update": True, "update_on_start": True},
"(auto_update, update_on_start)")]
for key, options, named in cases:
with self.subTest(key=key, options=options):
with self.assertRaises(Exception) as context:
self.build(licence_keys=key, **options)
message = str(context.exception)
self.assertIn(named, message)
self.assertIn("require a licence key", message)
self.assertIn("to False", message)

def test_the_pipeline_builder_needs_no_key_by_default(self):
pipeline = DeviceDetectionOnPremisePipelineBuilder(
data_file_path=data_file, usage_sharing=False).build()
self.addCleanup(pipeline.close)
self.assertIn("device", pipeline.flow_elements_list)

def test_an_update_url_other_than_the_distributor_needs_no_key(self):
engine = self.build(
data_update_url="https://updates.example.com/data.hash",
auto_update=True)
self.assertTrue(engine.data_file.auto_update)


if __name__ == "__main__":
unittest.main()
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
# *********************************************************************
# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
#
# This Original Work is licensed under the European Union Public Licence
# (EUPL) v.1.2 and is subject to its terms as set out below.
#
# If a copy of the EUPL was not distributed with this file, You can obtain
# one at https://opensource.org/licenses/EUPL-1.2.
#
# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
# amended by the European Commission) shall be deemed incompatible for
# the purposes of the Work and the provisions of the compatibility
# clause in Article 5 of the EUPL shall not apply.
#
# If using the Work as, or as part of, a network application, by
# including the attribution notice(s) required under Article 5 of the EUPL
# in the end user terms of the application under an appropriate heading,
# such notice(s) shall fulfill the requirements of that article.
# *********************************************************************

import os
import shutil
import tempfile
import threading
import unittest

from fiftyone_pipeline_engines.datafile_update_service import \
DataFileUpdateService, UpdateStatus
from fiftyone_devicedetection_onpremise.devicedetection_onpremise_pipelinebuilder import \
DeviceDetectionOnPremisePipelineBuilder

data_file = "../../cxx/data/device-detection/51Degrees-LiteV4.1.hash"

mobile_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"

# How often the watcher looks at the file in these tests.
WATCHER_INTERVAL_SECONDS = 0.5

# The longest the tests wait for the watcher to report the file and the
# engine to reload it. It is a guard that stops a test waiting for ever, and
# a reload that takes longer than it fails the test.
RELOAD_WAIT_SECONDS = 120


class WatcherWithoutLicenceKeyTests(unittest.TestCase):
"""!
The file system watcher downloads nothing, so it needs no licence key.
"""

def setUp(self):
folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder, True)
self.path = os.path.join(folder, "51Degrees-LiteV4.1.hash")
shutil.copyfile(data_file, self.path)

def build(self, **settings):
pipeline = DeviceDetectionOnPremisePipelineBuilder(
data_file_path=self.path,
usage_sharing=False,
performance_profile="MaxPerformance",
**settings).build()
self.addCleanup(pipeline.close)
return pipeline

def test_the_watcher_runs_with_no_licence_key(self):
pipeline = self.build(file_system_watcher=True)
watched = pipeline.get_element("device").data_file
self.assertTrue(watched.file_system_watcher)
self.assertFalse(watched.auto_update)
self.assertFalse(watched.update_on_start)

def test_a_replaced_file_is_loaded_with_no_licence_key(self):
service = DataFileUpdateService()
service.FILE_SYSTEM_WATCHER_INTERVAL_SECONDS = \
WATCHER_INTERVAL_SECONDS
statuses = []
reloaded = threading.Event()

def on_complete(status, watched):
statuses.append(status)
reloaded.set()

service.on_complete(on_complete)
pipeline = self.build(
file_system_watcher=True, data_file_update_service=service)

# Replace the file, and move its time on so that a file system
# that keeps times to the second still sees a change.
shutil.copyfile(data_file, self.path)
later = os.stat(self.path).st_mtime + 10
os.utime(self.path, (later, later))

self.assertTrue(
reloaded.wait(RELOAD_WAIT_SECONDS),
"The watcher did not report the replaced file.")
self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], statuses)

with pipeline.create_flowdata() as flowdata:
flowdata.evidence.add("header.user-agent", mobile_ua)
flowdata.process()
self.assertTrue(flowdata.device.ismobile.value())
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,10 @@ def __init__(self, settings = None, usage_sharing = True, **kwargs):
@type restricted_properties: list
@param restricted_properties: List of property keys to limit the engine to
@type licence_keys: string
@param licence_keys: licencekeys to use for the data file update service
@param licence_keys: licence keys to use for the data file update service.
Needed only when auto_update or update_on_start is on with the default
data_update_url, and the engine refuses to build without one then. The
file system watcher needs no key.
@type download: bool
@param download: whether to download the datafile or keep it in memory when it is returned from the datafile update service
@type performance_profile: string
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,16 @@
from .swig_data import SwigData
from .ipintelligence_datafile import IpIntelligenceDataFile

# The Distributor, which serves data files only to a licence key.
DISTRIBUTOR_URL = "https://distributor.51degrees.com/api/v2/download"

LICENCE_KEY_REQUIRED_FOR_UPDATES = (
"Some enabled features ({}) require a licence key but one has not been "
"supplied. Supply one with licence_keys, or set {} to False to build "
"without one. The file system watcher needs no licence key. A key can be "
"obtained from the 51Degrees website: "
"https://51degrees.com/pricing?utm_source=code&utm_medium=comment&utm_campaign=51degrees-python&utm_content=fiftyone_ipintelligence_onpremise-src-fiftyone_ipintelligence_onpremise-ipintelligence_onpremise.py&utm_term=licence-key-required-for-updates")


class IpIntelligenceOnPremise(Engine):
"""!
Expand All @@ -62,7 +72,7 @@ def __init__(
reuse_temp_file=False,
concurrency=multiprocessing.cpu_count(),
update_on_start=False,
data_update_url="https://distributor.51degrees.com/api/v2/download",
data_update_url=DISTRIBUTOR_URL,
data_update_use_url_formatter=True,
data_update_verify_md5=True,
data_update_product_type=None,
Expand All @@ -81,7 +91,10 @@ def __init__(
@type restricted_properties: list
@param restricted_properties: List of property keys to limit the engine to
@type licence_keys: string
@param licence_keys: licencekeys to use for the data file update service
@param licence_keys: licence keys to use for the data file update service.
Needed only when auto_update or update_on_start is on with the default
data_update_url, and the engine refuses to build without one then. The
file system watcher needs no key.
@type download: bool
@param download: whether to download the datafile or keep it in memory when it is returned from the datafile update service
@type performance_profile: string
Expand Down Expand Up @@ -121,6 +134,19 @@ def __init__(
if not data_file_path and not data:
raise Exception("data_file_path or data is required")

# Only polling and the update on start-up download from the
# Distributor, so only they need a licence key. The file system
# watcher needs none.
auto_update = bool(auto_update)
update_on_start = bool(update_on_start)
if data_update_url == DISTRIBUTOR_URL and not licence_keys:
need_key = [name for name, on in (
("auto_update", auto_update),
("update_on_start", update_on_start)) if on]
if need_key:
raise Exception(LICENCE_KEY_REQUIRED_FOR_UPDATES.format(
", ".join(need_key), " and ".join(need_key)))

if data_file_path:
if os.path.isfile(data_file_path) == False and not update_on_start:
raise Exception("There is no file at " + data_file_path)
Expand All @@ -132,9 +158,6 @@ def __init__(
else:
self.data = data

if not licence_keys and licence_keys != "":
raise Exception("licence key is required. A key can be obtained from the 51Degrees website: https://51degrees.com/pricing?utm_source=code&utm_medium=comment&utm_campaign=51degrees-python&utm_content=fiftyone_ipintelligence_onpremise-src-fiftyone_ipintelligence_onpremise-ipintelligence_onpremise.py&utm_term=licence-key-required. If you do not wish to use a key then you can specify an empty string, but this will cause automatic updates to be disabled.")

# Create SWIG wrapper vector for restricted properties and add

restricted_properties_vector = VectorStringSwig()
Expand Down Expand Up @@ -197,15 +220,6 @@ def __init__(
self.properties_list = properties_list
self.data_file_path = data_file_path

# Disable features that require a licence key if one was not supplied.

if licence_keys:
auto_update = auto_update and (len(licence_keys) > 0)
update_on_start = update_on_start and (len(licence_keys) > 0)
else:
auto_update = False
update_on_start = False

if auto_update or update_on_start or file_system_watcher:

# Construct DataFile
Expand Down
Loading
Loading