From 86f1ece6a983e28dcf9f1eb30381559e2e63fd4c Mon Sep 17 00:00:00 2001 From: Asaf Nave Date: Thu, 14 May 2026 13:07:33 +0300 Subject: [PATCH 1/2] SDK-363: Add edge.stats via HTTP Relay channel for device metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After filer firmware upgrades, the device-side metrics API (gw-api-handler) requires cookie-based session auth (_cteraSessionId_). The existing binary API channel (/devicecmdnew/) cannot carry HTTP headers by design, so auth always fails. This adds a new Stats command class that uses the portal's HTTP Relay channel (/devices//) — the same tunnel the Device UI uses — which forwards full raw HTTP including cookies end-to-end. - New cterasdk/edge/stats.py: Stats class with get() for all stat types and intervals, lazy SSO-based device authentication via the relay tunnel - Wire relay client in Clients.__init__ with lazy initialization (device name is only available after remote_command populates __dict__) - Expose edge.stats property on the Edge object (None for local edges) - 11 unit tests covering validation, SSO flow, and session reuse Ref: ESC-10493 Co-authored-by: Cursor --- cterasdk/edge/__init__.py | 1 + cterasdk/edge/stats.py | 65 +++++++++++++++++++++++++ cterasdk/objects/synchronous/edge.py | 19 +++++++- tests/ut/edge/test_stats.py | 71 ++++++++++++++++++++++++++++ 4 files changed, 154 insertions(+), 2 deletions(-) create mode 100644 cterasdk/edge/stats.py create mode 100644 tests/ut/edge/test_stats.py diff --git a/cterasdk/edge/__init__.py b/cterasdk/edge/__init__.py index be39ec2b..0b2d9c3c 100644 --- a/cterasdk/edge/__init__.py +++ b/cterasdk/edge/__init__.py @@ -30,6 +30,7 @@ 'shares', 'shell', 'smb', + 'stats', 'support', 'sync', 'syslog', diff --git a/cterasdk/edge/stats.py b/cterasdk/edge/stats.py new file mode 100644 index 00000000..7f1d553e --- /dev/null +++ b/cterasdk/edge/stats.py @@ -0,0 +1,65 @@ +import logging + +from ..common import parse_base_object_ref +from ..exceptions import CTERAException +from .base_command import BaseCommand + + +logger = logging.getLogger('cterasdk.edge') + + +VALID_STAT_TYPES = ('cpu', 'memory', 'cache', 'volume', 'connections', 'local_io', 'disk_io', 'cloud_io') +VALID_INTERVALS = ('hour', 'day', 'week', 'month', 'year', 'last') + + +class Stats(BaseCommand): + """ + Edge Filer statistics via the HTTP Relay channel. + + Uses the portal's RemoteDeviceServlet (/devices//) which forwards + all HTTP headers — including cookies — to the device, unlike the binary + API channel (/devicecmdnew/) which cannot carry cookies. + """ + + def __init__(self, edge, relay): + super().__init__(edge) + self._relay = relay + self._authenticated = False + + def get(self, stat_type, interval='hour'): + """ + Get device statistics + + :param str stat_type: Statistic type. + Options: ``cpu``, ``memory``, ``cache``, ``volume``, ``connections``, ``local_io``, ``disk_io``, ``cloud_io`` + :param str,optional interval: Time interval, defaults to ``hour``. + Options: ``hour``, ``day``, ``week``, ``month``, ``year``, ``last`` + :returns: Statistics data + """ + if stat_type not in VALID_STAT_TYPES: + raise ValueError(f'Invalid stat_type {stat_type!r}. Valid: {VALID_STAT_TYPES}') + if interval not in VALID_INTERVALS: + raise ValueError(f'Invalid interval {interval!r}. Valid: {VALID_INTERVALS}') + self._ensure_session() + return self._relay.get(f'/stats/{stat_type}', params={'interval': interval}) + + def _ensure_session(self): + """ + Authenticate to the device through the relay channel on first use. + + Retrieves an SSO token from the portal and uses it to log in to the + device via the relay tunnel. The device responds with a Set-Cookie + header (_cteraSessionId_) which the shared HTTP session stores + automatically, scoped to the relay path. + """ + if not self._authenticated: + Portal = self._edge._Portal # pylint: disable=protected-access + tenant = parse_base_object_ref(self._edge.portal).name + device_name = self._edge.name + logger.debug('Retrieving SSO ticket for relay session. %s', {'tenant': tenant, 'device': device_name}) + token = Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn') + if not token: + raise CTERAException('Failed to retrieve SSO ticket for relay session.') + self._relay.get('/ssologin', params={'ticket': token}) + self._authenticated = True + logger.debug('Relay session established. %s', {'tenant': tenant, 'device': device_name}) diff --git a/cterasdk/objects/synchronous/edge.py b/cterasdk/objects/synchronous/edge.py index dbe2c9ba..904fa82a 100644 --- a/cterasdk/objects/synchronous/edge.py +++ b/cterasdk/objects/synchronous/edge.py @@ -11,7 +11,7 @@ afp, aio, antivirus, array, audit, backup, cache, cli, config, connection, ctera_migrate, dedup, directoryservice, drive, files, firmware, ftp, groups, licenses, login, logs, mail, network, nfs, ntp, power, remote, rsync, ransom_protect, services, - shares, shell, smb, snmp, ssh, ssl, support, sync, syslog, tasks, telnet, + shares, shell, smb, snmp, ssh, ssl, stats, support, sync, syslog, tasks, telnet, timezone, users, volumes, ) @@ -19,6 +19,9 @@ class Clients: def __init__(self, edge, Portal): + self._edge = edge + self._Portal = Portal + self._stats = None if Portal: edge._Portal = Portal edge.default.close() @@ -29,6 +32,14 @@ def __init__(self, edge, Portal): self.api = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api')) self.io = IO(edge) + @property + def stats(self): + if self._stats is None and self._Portal: + relay_base = EndpointBuilder.new(f'{self._Portal.ctera.baseurl}/devices/{self._edge.name}') + relay = self._Portal.default.clone(clients.API, relay_base, authenticator=lambda *_: True) + self._stats = stats.Stats(self._edge, relay) + return self._stats + class IO: @@ -133,6 +144,10 @@ def api(self): def io(self): return self.clients.io + @property + def stats(self): + return self.clients.stats + @property def _session_id_key(self): return '_cteraSessionId_' @@ -164,5 +179,5 @@ def _omit_fields(self): return super()._omit_fields + ['afp', 'aio', 'array', 'audit', 'antivirus', 'backup', 'cache', 'cli', 'config', 'ctera_migrate', 'dedup', 'directoryservice', 'drive', 'files', 'firmware', 'ftp', 'groups', 'licenses', 'logs', 'mail', 'network', 'nfs', 'ntp', 'power', 'ransom_protect', 'rsync', 'services', 'shares', 'shell', - 'smb', 'snmp', 'ssh', 'ssl', 'support', 'sync', 'syslog', 'tasks', 'telnet', 'timezone', + 'smb', 'snmp', 'ssh', 'ssl', 'stats', 'support', 'sync', 'syslog', 'tasks', 'telnet', 'timezone', 'users', 'volumes'] diff --git a/tests/ut/edge/test_stats.py b/tests/ut/edge/test_stats.py new file mode 100644 index 00000000..20dd3af6 --- /dev/null +++ b/tests/ut/edge/test_stats.py @@ -0,0 +1,71 @@ +from unittest import mock + +from cterasdk.edge import stats +from cterasdk.exceptions import CTERAException +from tests.ut.edge import base_edge + + +class TestEdgeStats(base_edge.BaseEdgeTest): + + def setUp(self): + super().setUp() + self._relay = mock.MagicMock() + self._portal = mock.MagicMock() + self._filer._Portal = self._portal + self._filer.portal = 'objs/21//TeamPortal/test_tenant' + self._filer.name = 'my-filer' + self._sso_token = 'sso-token-abc123' + self._portal.api.execute.return_value = self._sso_token + self._stats = stats.Stats(self._filer, self._relay) + + def test_get_cpu_default_interval(self): + self._stats.get('cpu') + self._relay.get.assert_called_with('/stats/cpu', params={'interval': 'hour'}) + + def test_get_memory_with_interval(self): + self._stats.get('memory', interval='day') + self._relay.get.assert_called_with('/stats/memory', params={'interval': 'day'}) + + def test_get_all_stat_types(self): + for stat_type in stats.VALID_STAT_TYPES: + self._relay.reset_mock() + self._stats._authenticated = False + self._stats.get(stat_type, interval='hour') + self._relay.get.assert_called_with(f'/stats/{stat_type}', params={'interval': 'hour'}) + + def test_get_all_intervals(self): + for interval in stats.VALID_INTERVALS: + self._relay.reset_mock() + self._stats._authenticated = False + self._stats.get('cpu', interval=interval) + self._relay.get.assert_called_with('/stats/cpu', params={'interval': interval}) + + def test_invalid_stat_type_raises_value_error(self): + with self.assertRaises(ValueError): + self._stats.get('invalid_type') + + def test_invalid_interval_raises_value_error(self): + with self.assertRaises(ValueError): + self._stats.get('cpu', interval='invalid_interval') + + def test_ensure_session_retrieves_sso_token(self): + self._stats.get('cpu') + self._portal.api.execute.assert_called_once_with('/portals/test_tenant/devices/my-filer', 'singleSignOn') + + def test_ensure_session_calls_ssologin(self): + self._stats.get('cpu') + self._relay.get.assert_any_call('/ssologin', params={'ticket': self._sso_token}) + + def test_ensure_session_called_once(self): + self._stats.get('cpu') + self._stats.get('memory', interval='day') + self._portal.api.execute.assert_called_once() + + def test_ensure_session_raises_on_empty_token(self): + self._portal.api.execute.return_value = None + with self.assertRaises(CTERAException): + self._stats.get('cpu') + + def test_stats_not_available_locally(self): + local_filer = base_edge.Edge("") + self.assertIsNone(local_filer.stats) From 46165a02ea5bcc0c49323e77eee747613dbdd1d4 Mon Sep 17 00:00:00 2001 From: Asaf Nave Date: Thu, 14 May 2026 21:48:45 +0300 Subject: [PATCH 2/2] SDK-363: Switch all remote SDK calls to HTTP Relay channel Replace /devicecmdnew/{tenant}/{device}/ with /devices/{device}/ for all remote edge/drive API calls per the Jira AC. Add auto-SSO login on first api access through the relay channel so that cookie-based auth required by gw-api-handler is transparent. Simplify stats.py to use edge.api directly. Co-authored-by: Cursor --- cterasdk/core/remote.py | 4 +- cterasdk/edge/stats.py | 38 ++---------------- cterasdk/objects/synchronous/drive.py | 25 +++++++++++- cterasdk/objects/synchronous/edge.py | 34 +++++++++------- cterasdk/version.py | 24 +++++++++++ tests/ut/core/admin/test_remote.py | 33 +++++++++++++++ tests/ut/edge/test_stats.py | 58 ++++++--------------------- 7 files changed, 118 insertions(+), 98 deletions(-) create mode 100644 cterasdk/version.py diff --git a/cterasdk/core/remote.py b/cterasdk/core/remote.py index 432f7c35..e96422e3 100644 --- a/cterasdk/core/remote.py +++ b/cterasdk/core/remote.py @@ -1,11 +1,9 @@ from .enum import DeviceType from ..objects.synchronous import edge, drive -from ..common import parse_base_object_ref def remote_command(Portal, device): - tenant = parse_base_object_ref(device.portal).name - base = f'{Portal.ctera.baseurl}/devicecmdnew/{tenant}/{device.name}' + base = f'{Portal.ctera.baseurl}/devices/{device.name}' ManagedDevice = None if device.deviceType in DeviceType.Gateways: diff --git a/cterasdk/edge/stats.py b/cterasdk/edge/stats.py index 7f1d553e..10b6d870 100644 --- a/cterasdk/edge/stats.py +++ b/cterasdk/edge/stats.py @@ -1,7 +1,5 @@ import logging -from ..common import parse_base_object_ref -from ..exceptions import CTERAException from .base_command import BaseCommand @@ -14,18 +12,12 @@ class Stats(BaseCommand): """ - Edge Filer statistics via the HTTP Relay channel. + Edge Filer statistics retrieved via the HTTP Relay channel. - Uses the portal's RemoteDeviceServlet (/devices//) which forwards - all HTTP headers — including cookies — to the device, unlike the binary - API channel (/devicecmdnew/) which cannot carry cookies. + Valid stat types: cpu, memory, cache, volume, connections, local_io, disk_io, cloud_io + Valid intervals: hour, day, week, month, year, last """ - def __init__(self, edge, relay): - super().__init__(edge) - self._relay = relay - self._authenticated = False - def get(self, stat_type, interval='hour'): """ Get device statistics @@ -40,26 +32,4 @@ def get(self, stat_type, interval='hour'): raise ValueError(f'Invalid stat_type {stat_type!r}. Valid: {VALID_STAT_TYPES}') if interval not in VALID_INTERVALS: raise ValueError(f'Invalid interval {interval!r}. Valid: {VALID_INTERVALS}') - self._ensure_session() - return self._relay.get(f'/stats/{stat_type}', params={'interval': interval}) - - def _ensure_session(self): - """ - Authenticate to the device through the relay channel on first use. - - Retrieves an SSO token from the portal and uses it to log in to the - device via the relay tunnel. The device responds with a Set-Cookie - header (_cteraSessionId_) which the shared HTTP session stores - automatically, scoped to the relay path. - """ - if not self._authenticated: - Portal = self._edge._Portal # pylint: disable=protected-access - tenant = parse_base_object_ref(self._edge.portal).name - device_name = self._edge.name - logger.debug('Retrieving SSO ticket for relay session. %s', {'tenant': tenant, 'device': device_name}) - token = Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn') - if not token: - raise CTERAException('Failed to retrieve SSO ticket for relay session.') - self._relay.get('/ssologin', params={'ticket': token}) - self._authenticated = True - logger.debug('Relay session established. %s', {'tenant': tenant, 'device': device_name}) + return self._edge.api.get(f'/stats/{stat_type}', params={'interval': interval}) diff --git a/cterasdk/objects/synchronous/drive.py b/cterasdk/objects/synchronous/drive.py index 2ab59a78..974877aa 100644 --- a/cterasdk/objects/synchronous/drive.py +++ b/cterasdk/objects/synchronous/drive.py @@ -1,21 +1,42 @@ +import logging + import cterasdk.settings from ...clients import clients from ..services import Management from ..endpoints import EndpointBuilder +from ...common import parse_base_object_ref from ...lib.session.edge import Session from ...edge import backup, cli, logs, services, support, sync +logger = logging.getLogger('cterasdk.drive') + + class Clients: def __init__(self, drive, Portal): + self._drive = drive + self._Portal = Portal + self._authenticated = False if Portal: drive._Portal = Portal drive.default.close() drive._ctera_session.start_remote_session(Portal.session()) - self.api = Portal.default.clone(clients.API, EndpointBuilder.new(drive.base), authenticator=lambda *_: True) + self._api = Portal.default.clone(clients.API, EndpointBuilder.new(drive.base), authenticator=lambda *_: True) else: - self.api = drive.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api')) + self._api = drive.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api')) + + @property + def api(self): + if self._Portal and not self._authenticated: + tenant = parse_base_object_ref(self._drive.portal).name + device_name = self._drive.name + logger.debug('Auto-SSO login via relay channel. %s', {'tenant': tenant, 'device': device_name}) + token = self._Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn') + if token: + self._api.get('/ssologin', params={'ticket': token}) + self._authenticated = True + return self._api class Drive(Management): diff --git a/cterasdk/objects/synchronous/edge.py b/cterasdk/objects/synchronous/edge.py index 904fa82a..b3e66da0 100644 --- a/cterasdk/objects/synchronous/edge.py +++ b/cterasdk/objects/synchronous/edge.py @@ -1,9 +1,11 @@ +import logging + import cterasdk.settings from ...clients import clients from ..services import Management from ..endpoints import EndpointBuilder from .. import authenticators -from ...common import modules +from ...common import modules, parse_base_object_ref from ...lib.session.edge import Session @@ -16,29 +18,36 @@ ) +logger = logging.getLogger('cterasdk.edge') + + class Clients: def __init__(self, edge, Portal): self._edge = edge self._Portal = Portal - self._stats = None + self._authenticated = False if Portal: edge._Portal = Portal edge.default.close() edge._ctera_session.start_remote_session(Portal.session()) - self.api = Portal.default.clone(clients.API, EndpointBuilder.new(edge.base), authenticator=lambda *_: True) + self._api = Portal.default.clone(clients.API, EndpointBuilder.new(edge.base), authenticator=lambda *_: True) else: self.migrate = edge.default.clone(clients.Migrate, EndpointBuilder.new(edge.base, '/migration/rest/v1')) - self.api = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api')) + self._api = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api')) self.io = IO(edge) @property - def stats(self): - if self._stats is None and self._Portal: - relay_base = EndpointBuilder.new(f'{self._Portal.ctera.baseurl}/devices/{self._edge.name}') - relay = self._Portal.default.clone(clients.API, relay_base, authenticator=lambda *_: True) - self._stats = stats.Stats(self._edge, relay) - return self._stats + def api(self): + if self._Portal and not self._authenticated: + tenant = parse_base_object_ref(self._edge.portal).name + device_name = self._edge.name + logger.debug('Auto-SSO login via relay channel. %s', {'tenant': tenant, 'device': device_name}) + token = self._Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn') + if token: + self._api.get('/ssologin', params={'ticket': token}) + self._authenticated = True + return self._api class IO: @@ -117,6 +126,7 @@ def __init__(self, host=None, port=None, https=True, Portal=None, *, base=None): self.shell = shell.Shell(self) self.smb = smb.SMB(self) self.snmp = snmp.SNMP(self) + self.stats = stats.Stats(self) self.ssh = ssh.SSH(self) self.ssl = modules.initialize(ssl.SSLModule, self) self.support = support.Support(self) @@ -144,10 +154,6 @@ def api(self): def io(self): return self.clients.io - @property - def stats(self): - return self.clients.stats - @property def _session_id_key(self): return '_cteraSessionId_' diff --git a/cterasdk/version.py b/cterasdk/version.py new file mode 100644 index 00000000..62e8eded --- /dev/null +++ b/cterasdk/version.py @@ -0,0 +1,24 @@ +# file generated by vcs-versioning +# don't change, don't track in version control +from __future__ import annotations + +__all__ = [ + "__version__", + "__version_tuple__", + "version", + "version_tuple", + "__commit_id__", + "commit_id", +] + +version: str +__version__: str +__version_tuple__: tuple[int | str, ...] +version_tuple: tuple[int | str, ...] +commit_id: str | None +__commit_id__: str | None + +__version__ = version = '2.20.35.dev0+g99746a14c.d20260514' +__version_tuple__ = version_tuple = (2, 20, 35, 'dev0', 'g99746a14c.d20260514') + +__commit_id__ = commit_id = 'g99746a14c' diff --git a/tests/ut/core/admin/test_remote.py b/tests/ut/core/admin/test_remote.py index 12e05d55..65544686 100644 --- a/tests/ut/core/admin/test_remote.py +++ b/tests/ut/core/admin/test_remote.py @@ -4,6 +4,7 @@ from cterasdk.core import devices from cterasdk.common import Object from cterasdk.objects import Edge, Drive +from cterasdk.objects.synchronous.edge import Clients as EdgeClients from tests.ut.core.admin import base_admin @@ -73,6 +74,38 @@ def _create_device_param(name, portal, device_type, remote_access_url): param.remoteAccessUrl = remote_access_url return param + def test_auto_sso_on_first_api_access(self): + """Verify that first access to edge.api triggers SSO login via relay channel.""" + remote_session = self.patch_call("cterasdk.lib.session.edge.Session.start_remote_session") + remote_session.return_value = munch.Munch({'account': munch.Munch({'name': 'mickey', 'tenant': 'tenant'})}) + get_multi_response = TestCoreRemote._create_device_param(self._device_name, self._device_portal, + 'vGateway', self._device_remote_access_url) + self._init_global_admin(get_multi_response=get_multi_response, execute_response=self._sso_ticket) + self._activate_portal_session() + device = devices.Devices(self._global_admin).device(self._device_name) + device._ctera_clients._api = mock.MagicMock() + _ = device.api + self._global_admin.api.execute.assert_called_once_with( + f'/portals/{self._tenant_name}/devices/{self._device_name}', 'singleSignOn') + device._ctera_clients._api.get.assert_called_once_with('/ssologin', params={'ticket': self._sso_ticket}) + + def test_auto_sso_not_repeated_on_subsequent_api_access(self): + """Verify that subsequent api accesses do not re-trigger SSO.""" + remote_session = self.patch_call("cterasdk.lib.session.edge.Session.start_remote_session") + remote_session.return_value = munch.Munch({'account': munch.Munch({'name': 'mickey', 'tenant': 'tenant'})}) + get_multi_response = TestCoreRemote._create_device_param(self._device_name, self._device_portal, + 'vGateway', self._device_remote_access_url) + self._init_global_admin(get_multi_response=get_multi_response, execute_response=self._sso_ticket) + self._activate_portal_session() + device = devices.Devices(self._global_admin).device(self._device_name) + device._ctera_clients._api = mock.MagicMock() + _ = device.api + _ = device.api + _ = device.api + self._global_admin.api.execute.assert_called_once_with( + f'/portals/{self._tenant_name}/devices/{self._device_name}', 'singleSignOn') + device._ctera_clients._api.get.assert_called_once_with('/ssologin', params={'ticket': self._sso_ticket}) + @staticmethod def _create_current_session_object(): session = Object() diff --git a/tests/ut/edge/test_stats.py b/tests/ut/edge/test_stats.py index 20dd3af6..1b5f6456 100644 --- a/tests/ut/edge/test_stats.py +++ b/tests/ut/edge/test_stats.py @@ -1,7 +1,6 @@ from unittest import mock from cterasdk.edge import stats -from cterasdk.exceptions import CTERAException from tests.ut.edge import base_edge @@ -9,63 +8,32 @@ class TestEdgeStats(base_edge.BaseEdgeTest): def setUp(self): super().setUp() - self._relay = mock.MagicMock() - self._portal = mock.MagicMock() - self._filer._Portal = self._portal - self._filer.portal = 'objs/21//TeamPortal/test_tenant' - self._filer.name = 'my-filer' - self._sso_token = 'sso-token-abc123' - self._portal.api.execute.return_value = self._sso_token - self._stats = stats.Stats(self._filer, self._relay) + self._init_filer() def test_get_cpu_default_interval(self): - self._stats.get('cpu') - self._relay.get.assert_called_with('/stats/cpu', params={'interval': 'hour'}) + stats.Stats(self._filer).get('cpu') + self._filer.api.get.assert_called_with('/stats/cpu', params={'interval': 'hour'}) def test_get_memory_with_interval(self): - self._stats.get('memory', interval='day') - self._relay.get.assert_called_with('/stats/memory', params={'interval': 'day'}) + stats.Stats(self._filer).get('memory', interval='day') + self._filer.api.get.assert_called_with('/stats/memory', params={'interval': 'day'}) def test_get_all_stat_types(self): for stat_type in stats.VALID_STAT_TYPES: - self._relay.reset_mock() - self._stats._authenticated = False - self._stats.get(stat_type, interval='hour') - self._relay.get.assert_called_with(f'/stats/{stat_type}', params={'interval': 'hour'}) + self._filer.api.get.reset_mock() + stats.Stats(self._filer).get(stat_type, interval='hour') + self._filer.api.get.assert_called_with(f'/stats/{stat_type}', params={'interval': 'hour'}) def test_get_all_intervals(self): for interval in stats.VALID_INTERVALS: - self._relay.reset_mock() - self._stats._authenticated = False - self._stats.get('cpu', interval=interval) - self._relay.get.assert_called_with('/stats/cpu', params={'interval': interval}) + self._filer.api.get.reset_mock() + stats.Stats(self._filer).get('cpu', interval=interval) + self._filer.api.get.assert_called_with('/stats/cpu', params={'interval': interval}) def test_invalid_stat_type_raises_value_error(self): with self.assertRaises(ValueError): - self._stats.get('invalid_type') + stats.Stats(self._filer).get('invalid_type') def test_invalid_interval_raises_value_error(self): with self.assertRaises(ValueError): - self._stats.get('cpu', interval='invalid_interval') - - def test_ensure_session_retrieves_sso_token(self): - self._stats.get('cpu') - self._portal.api.execute.assert_called_once_with('/portals/test_tenant/devices/my-filer', 'singleSignOn') - - def test_ensure_session_calls_ssologin(self): - self._stats.get('cpu') - self._relay.get.assert_any_call('/ssologin', params={'ticket': self._sso_token}) - - def test_ensure_session_called_once(self): - self._stats.get('cpu') - self._stats.get('memory', interval='day') - self._portal.api.execute.assert_called_once() - - def test_ensure_session_raises_on_empty_token(self): - self._portal.api.execute.return_value = None - with self.assertRaises(CTERAException): - self._stats.get('cpu') - - def test_stats_not_available_locally(self): - local_filer = base_edge.Edge("") - self.assertIsNone(local_filer.stats) + stats.Stats(self._filer).get('cpu', interval='invalid_interval')