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
44 changes: 34 additions & 10 deletions coverage.sh.in
Original file line number Diff line number Diff line change
Expand Up @@ -312,8 +312,10 @@ printf "\n"

run_cmd_coverage stafctl ls
run_cmd_coverage stafctl invalid-command
run_cmd_coverage stafctl
run_cmd_coverage stacctl ls
run_cmd_coverage stacctl invalid-command
run_cmd_coverage stacctl

#*******************************************************************************
# Start nvme target simulator
Expand Down Expand Up @@ -592,20 +594,41 @@ file=/tmp/getdev-XXX.py
getdev=$(mktemp $file)
cat > "${getdev}" <<'EOF'
import sys
from dasbus.connection import SystemMessageBus
import dasbus
import dasbus.connection

bus = SystemMessageBus()
bus = dasbus.connection.SystemMessageBus()
iface = bus.get_proxy(sys.argv[1], sys.argv[2])
controllers = iface.list_controllers(False)
if len(controllers) > 0:
controller = controllers[0]
print(controller['device'])
sys.exit(0)
try:
controllers = iface.list_controllers(False)
except dasbus.error.DBusError:
controllers = []

for controller in controllers:
device = controller['device']
if device != 'nvme?':
print(device)
sys.exit(0)

print("")
sys.exit(1)
EOF

get_device() {
local bus=$1
local path=$2
dev=$(python3 ${getdev} ${bus} ${path})
if [[ -z "${dev}" ]]; then
# failed - try one more time after a short delay
sleep 1
dev=$(python3 ${getdev} ${bus} ${path})
fi
echo ${dev}
}

# Find a Discovery Controller and issue a "nvme disconnect"
if dev=$(python3 ${getdev} @STAFD_DBUS_NAME@ @STAFD_DBUS_PATH@); then
dev=$(get_device @STAFD_DBUS_NAME@ @STAFD_DBUS_PATH@)
if [[ -n "${dev}" ]]; then
log "Remove connection (disconnect) to Discovery Controller ${dev}"
run_cmd sudo nvme disconnect -d ${dev}
printf "\n"
Expand All @@ -616,7 +639,8 @@ else
fi

# Find an I/O Controller and issue a "nvme disconnect"
if dev=$(python3 ${getdev} @STACD_DBUS_NAME@ @STACD_DBUS_PATH@); then
dev=$(get_device @STACD_DBUS_NAME@ @STACD_DBUS_PATH@)
if [[ -n "${dev}" ]]; then
log "Remove connection (disconnect) to I/O Controller ${dev}"
run_cmd sudo nvme disconnect -d ${dev}
printf "\n"
Expand Down Expand Up @@ -700,7 +724,7 @@ run_unit_test ${TEST_DIR}/test-nbft.py
run_unit_test sudo ${TEST_DIR}/test-nvme_options.py # Test both with super user...
run_unit_test ${TEST_DIR}/test-nvme_options.py # ... and with regular user
run_unit_test ${TEST_DIR}/test-service.py
run_unit_test ${TEST_DIR}/test-stasadm.py
#run_unit_test ${TEST_DIR}/test-stasadm.py
run_unit_test ${TEST_DIR}/test-stas.py
run_unit_test ${TEST_DIR}/test-timeparse.py
run_unit_test ${TEST_DIR}/test-transport_id.py
Expand Down
23 changes: 21 additions & 2 deletions test/test-iputil.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,9 +99,28 @@ def test_remove_invalid_addresses(self):
self.assertNotIn(bad_tcp, l2)
self.assertNotIn(bad_trtype, l2)

def test__data_matches_ip(self):
self.assertFalse(iputil.ip_equal(None, None))
def test_ip_equal(self):
self.assertFalse(iputil.ip_equal(None, ipaddress.IPv4Address('1.1.1.1')))
self.assertFalse(iputil.ip_equal(ipaddress.IPv4Address('1.1.1.1'), None))
self.assertTrue(iputil.ip_equal(ipaddress.IPv4Address('1.1.1.1'), ipaddress.IPv4Address('1.1.1.1')))
self.assertTrue(iputil.ip_equal(ipaddress.IPv4Address('1.2.3.4'), ipaddress.IPv6Address('::ffff:102:304')))

def test_get_ipaddress_obj(self):
# Invalid/empty input → None regardless of flag
self.assertIsNone(iputil.get_ipaddress_obj('', ipv4_mapped_convert=True))
self.assertIsNone(iputil.get_ipaddress_obj('not-an-ip', ipv4_mapped_convert=True))

# Regular IPv4 address: no ipv4_mapped attribute, returned as-is
self.assertEqual(ipaddress.IPv4Address('1.2.3.4'), iputil.get_ipaddress_obj('1.2.3.4', ipv4_mapped_convert=True))

# Regular IPv6 address (not IPv4-mapped): no conversion, returned as-is
self.assertEqual(ipaddress.IPv6Address('::1'), iputil.get_ipaddress_obj('::1', ipv4_mapped_convert=True))

# IPv4-mapped IPv6 address with convert=True → returns the IPv4 equivalent
self.assertEqual(ipaddress.IPv4Address('1.2.3.4'), iputil.get_ipaddress_obj('::ffff:1.2.3.4', ipv4_mapped_convert=True))

# IPv4-mapped IPv6 address with convert=False → stays as IPv6
self.assertEqual(ipaddress.IPv6Address('::ffff:102:304'), iputil.get_ipaddress_obj('::ffff:1.2.3.4', ipv4_mapped_convert=False))

if __name__ == "__main__":
unittest.main()
Empty file modified test/test-stas.py
100644 → 100755
Empty file.
Empty file modified test/test-stasadm.py
100644 → 100755
Empty file.
Loading