-
Notifications
You must be signed in to change notification settings - Fork 85
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add dbus service to read file stat (#142)
* add dbus file stat * update
- Loading branch information
1 parent
ca6b3cd
commit 1891b0a
Showing
3 changed files
with
105 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,45 @@ | ||
"""File stat handler""" | ||
|
||
from host_modules import host_service | ||
import subprocess | ||
|
||
MOD_NAME = 'file' | ||
EXIT_FAILURE = 1 | ||
|
||
import os | ||
|
||
class FileService(host_service.HostModule): | ||
""" | ||
Dbus endpoint that executes the file command | ||
""" | ||
@host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='ia{ss}') | ||
def get_file_stat(self, path): | ||
if not path: | ||
return EXIT_FAILURE, {'error': 'Dbus get_file_stat called with no path specified'} | ||
|
||
try: | ||
file_stat = os.stat(path) | ||
|
||
# Get last modified time in nanoseconds since epoch | ||
last_modified = int(file_stat.st_mtime * 1e9) # Convert seconds to nanoseconds | ||
|
||
# Get permissions in octal format | ||
permissions = oct(file_stat.st_mode)[-3:] | ||
|
||
# Get file size in bytes | ||
size = file_stat.st_size | ||
|
||
# Get current umask | ||
current_umask = os.umask(0) | ||
os.umask(current_umask) # Reset umask to previous value | ||
|
||
return 0, { | ||
'path': path, | ||
'last_modified': str(last_modified), # Converting to string to maintain consistency | ||
'permissions': permissions, | ||
'size': str(size), # Converting to string to maintain consistency | ||
'umask': oct(current_umask)[-3:] | ||
} | ||
|
||
except Exception as e: | ||
return EXIT_FAILURE, {'error': str(e)} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
import sys | ||
import os | ||
import pytest | ||
from unittest import mock | ||
from host_modules import file_service | ||
|
||
class TestFileService(object): | ||
@mock.patch("dbus.SystemBus") | ||
@mock.patch("dbus.service.BusName") | ||
@mock.patch("dbus.service.Object.__init__") | ||
@mock.patch("os.stat") | ||
@mock.patch("os.umask") | ||
def test_get_file_stat_valid(self, mock_umask, mock_stat, MockInit, MockBusName, MockSystemBus): | ||
mock_stat_result = mock.Mock() | ||
mock_stat_result.st_mtime = 1609459200.0 # 2021-01-01 00:00:00 in nanoseconds | ||
mock_stat_result.st_mode = 0o100644 # Regular file with permissions | ||
mock_stat_result.st_size = 1024 | ||
mock_stat.return_value = mock_stat_result | ||
|
||
mock_umask.return_value = 0o022 # Default umask | ||
|
||
file_service_stub = file_service.FileService(file_service.MOD_NAME) | ||
path = "/valid/path" | ||
ret, msg = file_service_stub.get_file_stat(path) | ||
|
||
assert ret == 0 | ||
assert msg['path'] == path | ||
assert msg['last_modified'] == "1609459200000000000" | ||
assert msg['permissions'] == "644" | ||
assert msg['size'] == "1024" | ||
assert msg['umask'] == "o22" | ||
|
||
@mock.patch("dbus.SystemBus") | ||
@mock.patch("dbus.service.BusName") | ||
@mock.patch("dbus.service.Object.__init__") | ||
@mock.patch("os.stat") | ||
def test_get_file_stat_invalid_path(self, mock_stat, MockInit, MockBusName, MockSystemBus): | ||
mock_stat.side_effect = FileNotFoundError("[Errno 2] No such file or directory") | ||
|
||
file_service_stub = file_service.FileService(file_service.MOD_NAME) | ||
path = "/invalid/path" | ||
ret, msg = file_service_stub.get_file_stat(path) | ||
|
||
assert ret == 1 | ||
assert 'error' in msg | ||
assert "No such file or directory" in msg['error'] | ||
|
||
@mock.patch("dbus.SystemBus") | ||
@mock.patch("dbus.service.BusName") | ||
@mock.patch("dbus.service.Object.__init__") | ||
def test_get_file_stat_empty_path(self, MockInit, MockBusName, MockSystemBus): | ||
file_service_stub = file_service.FileService(file_service.MOD_NAME) | ||
path = "" | ||
ret, msg = file_service_stub.get_file_stat(path) | ||
|
||
assert ret == 1 | ||
assert "Dbus get_file_stat called with no path specified" in msg['error'] |