Merge branch 'master' into fix/batch_append_update_bust_version_chain

Author: phoebusm

python/arcticdb/storage_fixtures/azure.py

@@ -13,6 +13,7 @@
 import stat
 import tempfile
 import uuid
+import requests
 from typing import TYPE_CHECKING, Optional, Union
 from tempfile import mkdtemp
 
@@ -214,8 +215,25 @@ def _safe_enter(self):
             process_start_cmd=args,
             cwd=self.working_dir,
         )
+        if not self.ssl_test_support:
+            # Do not verify when ssl test support because of need of proper certs
+            assert self.is_azurite(self.endpoint_root), f"Azurite not started at: {self.endpoint_root}"
         return self
 
+    def is_azurite(self, url: str, timeout: int = 60):
+        try:
+            response = requests.get(url, timeout=timeout)
+            headers = response.headers
+
+            # Check for Azurite-specific headers
+            server_header = headers.get("Server", "").lower()
+            has_azurite_headers = "x-ms-request-id" in str(headers) and "azurite" in server_header
+
+            return has_azurite_headers
+
+        except requests.RequestException:
+            return False
+
     def __exit__(self, exc_type, exc_value, traceback):
         with handle_cleanup_exception(self, "process", consequence="Subsequent file deletion may also fail. "):
             GracefulProcessUtils.terminate(self._p)

python/arcticdb/storage_fixtures/mongo.py

@@ -117,7 +117,7 @@ class ManagedMongoDBServer(StorageFixtureFactory):
 
     def __init__(self, data_dir: Optional[str] = None, port=0, executable="mongod"):
         self._data_dir = data_dir or tempfile.mkdtemp("ManagedMongoDBServer")
-        self._port = port or get_ephemeral_port(5)
+        self._port = port or get_ephemeral_port(7)
         self._executable = executable
         self._client = None
 

python/arcticdb/storage_fixtures/s3.py

@@ -13,6 +13,7 @@
 import re
 import sys
 import platform
+import pprint
 from tempfile import mkdtemp
 from urllib.parse import urlparse
 import boto3
@@ -688,13 +689,20 @@ def __call__(self, environ, start_response):
             else:
                 self._reqs_till_rate_limit -= 1
 
+            # Lets add ability to identify type as S3
+            if "/whoami" in path_info:
+                start_response("200 OK", [("Content-Type", "text/plain")])
+                return [b"Moto AWS S3"]
+
         return super().__call__(environ, start_response)
 
 
 class GcpHostDispatcherApplication(HostDispatcherApplication):
     """GCP's S3 implementation does not have batch delete."""
 
     def __call__(self, environ, start_response):
+        path_info: bytes = environ.get("PATH_INFO", "")
+
         if environ["REQUEST_METHOD"] == "POST" and environ["QUERY_STRING"] == "delete":
             response_body = (
                 b'<?xml version="1.0" encoding="UTF-8"?>'
@@ -708,6 +716,12 @@ def __call__(self, environ, start_response):
                 "501 Not Implemented", [("Content-Type", "text/xml"), ("Content-Length", str(len(response_body)))]
             )
             return [response_body]
+
+        # Lets add ability to identify type as GCP
+        if "/whoami" in path_info:
+            start_response("200 OK", [("Content-Type", "text/plain")])
+            return [b"Moto GCP"]
+
         return super().__call__(environ, start_response)
 
 
@@ -731,6 +745,20 @@ def run_gcp_server(port, key_file, cert_file):
     )
 
 
+def is_server_type(url: str, server_type: str):
+    """Check if a server is of certain type.
+
+    /whoami url is added to Moto* objects to identify GCP or S3"""
+    try:
+        response = requests.get(url, verify=False)
+        if response.status_code == 200 and server_type in response.text:
+            return True
+    except Exception as e:
+        logger.error(f"Error during server type check: {e}")
+    logger.error(f"Was not of expected type: status code {response.status_code}, text: {response.text}")
+    return False
+
+
 def create_bucket(s3_client, bucket_name, max_retries=15):
     for i in range(max_retries):
         try:
@@ -741,6 +769,9 @@ def create_bucket(s3_client, bucket_name, max_retries=15):
                 raise
             logger.warning(f"S3 create bucket failed. Retry {1}/{max_retries}")
             time.sleep(1)
+        except Exception as e:
+            logger.error(f"create_bucket - Error: {e.response['Error']['Message']}")
+            pprint.pprint(e.response)
 
 
 class MotoS3StorageFixtureFactory(BaseS3StorageFixtureFactory):
@@ -791,10 +822,19 @@ def bucket_name(self, bucket_type="s3"):
         # We need the unique_id because we have tests that are creating the factory directly
         # and not using the fixtures
         # so this guarantees a unique bucket name
-        return f"test_{bucket_type}_bucket_{self.unique_id}_{self._bucket_id}"
+        return f"test-{bucket_type}-bucket-{self.unique_id}-{self._bucket_id}"
 
-    def _start_server(self):
-        port = self.port = get_ephemeral_port(2)
+    def is_server_type(url: str, server_type: str):
+        try:
+            response = requests.get(url, verify=False)
+            if response.status_code == 200 and server_type in response.text:
+                return True
+        except Exception:
+            pass
+        return False
+
+    def _start_server(self, seed=2):
+        port = self.port = get_ephemeral_port(seed)
         self.endpoint = f"{self.http_protocol}://{self.host}:{port}"
         self.working_dir = mkdtemp(suffix="MotoS3StorageFixtureFactory")
         self._iam_endpoint = f"{self.http_protocol}://localhost:{port}"
@@ -826,17 +866,22 @@ def _start_server(self):
         # There is a problem with the performance of the socket module in the MacOS 15 GH runners - https://github.com/actions/runner-images/issues/12162
         # Due to this, we need to wait for the server to come up for a longer time
         wait_for_server_to_come_up(self.endpoint, "moto", self._p, timeout=240)
+        assert is_server_type(self.endpoint + "/whoami", "S3"), "The server has not identified as S3"
 
     def _safe_enter(self):
-        for _ in range(3):  # For unknown reason, Moto, when running in pytest-xdist, will randomly fail to start
+        for i in range(5):  # For unknown reason, Moto, when running in pytest-xdist, will randomly fail to start
             try:
-                self._start_server()
+                logger.info(f"Attempt to start server - {i}")
+                self._start_server(2 + i)
+                self._s3_admin = self._boto(service="s3", key=self.default_key)
+                logger.info(f"Moto S3 STARTED!!! on port {self.port}")
                 break
             except AssertionError as e:  # Thrown by wait_for_server_to_come_up
                 sys.stderr.write(repr(e))
                 GracefulProcessUtils.terminate(self._p)
+            except Exception as e:
+                logger.error(f"Error during startup of Moto S3. Trying again. Error: {e}")
 
-        self._s3_admin = self._boto(service="s3", key=self.default_key)
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
@@ -928,8 +973,8 @@ def create_fixture(self) -> NfsS3Bucket:
 
 
 class MotoGcpS3StorageFixtureFactory(MotoS3StorageFixtureFactory):
-    def _start_server(self):
-        port = self.port = get_ephemeral_port(3)
+    def _start_server(self, seed=20):
+        port = self.port = get_ephemeral_port(seed)
         self.endpoint = f"{self.http_protocol}://{self.host}:{port}"
         self.working_dir = mkdtemp(suffix="MotoGcpS3StorageFixtureFactory")
         self._iam_endpoint = f"{self.http_protocol}://localhost:{port}"
@@ -961,6 +1006,7 @@ def _start_server(self):
         # There is a problem with the performance of the socket module in the MacOS 15 GH runners - https://github.com/actions/runner-images/issues/12162
         # Due to this, we need to wait for the server to come up for a longer time
         wait_for_server_to_come_up(self.endpoint, "moto", self._p, timeout=240)
+        assert is_server_type(self.endpoint + "/whoami", "GCP"), "The server has not identified as GCP"
 
     def create_fixture(self) -> GcpS3Bucket:
         bucket = self.bucket_name("gcp")

python/tests/integration/arcticdb/test_s3.py

@@ -196,7 +196,6 @@ def test_wrapped_s3_storage(lib_name, wrapped_s3_storage_bucket):
     lib.write("s", data=create_df())
 
 
-@SKIP_CONDA_MARK  # issue with fixture init will be fixed in https://github.com/man-group/ArcticDB/issues/2640
 def test_library_get_key_path(lib_name, s3_and_nfs_storage_bucket, test_prefix):
     lib = s3_and_nfs_storage_bucket.create_version_store_factory(lib_name)()
     lib.write("s", data=create_df())

python/tests/integration/arcticdb/version_store/test_basic_version_store.py

@@ -31,6 +31,7 @@
 )
 from arcticdb import QueryBuilder
 from arcticdb.flattener import Flattener
+from arcticdb.util.utils import generate_random_numpy_array, generate_random_series
 from arcticdb.version_store import NativeVersionStore
 from arcticdb.version_store._store import VersionedItem
 from arcticdb_ext.exceptions import _ArcticLegacyCompatibilityException, StorageException
@@ -1160,6 +1161,37 @@ def test_update_times(basic_store):
     assert update_times_versioned[0] < update_times_versioned[1] < update_times_versioned[2]
 
 
+@pytest.mark.storage
+def test_update_time(basic_store):
+    lib: NativeVersionStore = basic_store
+
+    nparr = generate_random_numpy_array(50, np.float32, seed=None)
+    series = generate_random_series(np.uint64, 50, "numbers")
+    df = pd.DataFrame(data={"col1": np.arange(10)}, index=pd.date_range(pd.Timestamp(0), periods=10))
+    lib.write("sym1", nparr)
+    lib.write("sym1", series)
+    lib.snapshot("snap")
+    lib.write("sym1", df)
+
+    # Negative number for as_of works as expected (dataframes)
+    assert lib.update_time("sym1") == lib.update_time("sym1", -1) == lib.update_time("sym1", 2)
+    # Snapshots are accepted (series)
+    assert lib.update_time("sym1", 1) == lib.update_time("sym1", -2) == lib.update_time("sym1", "snap")
+    # and now check for np array
+    assert lib.update_time("sym1", 0) == lib.update_time("sym1", -3)
+
+    # Times are ordered
+    assert lib.update_time("sym1") > lib.update_time("sym1", 1) > lib.update_time("sym1", 0)
+
+    # Correct exception thrown if symbol does not exist
+    with pytest.raises(NoDataFoundException):
+        lib.update_time("sym12")
+
+    # Correct exception thrown if version does not exist
+    with pytest.raises(NoDataFoundException):
+        lib.update_time("sym1", 11)
+
+
 @pytest.mark.storage
 @pytest.mark.parametrize(
     "index_names",
@@ -2556,6 +2588,7 @@ def test_batch_restore_version_mixed_as_ofs(lmdb_version_store):
     second_ts = pd.Timestamp(2000)
     ManualClockVersionStore.time = second_ts.value
     lib.batch_write(syms, second_data, second_metadata)
+    lib.snapshot("snap")
 
     third_ts = pd.Timestamp(3000)
     ManualClockVersionStore.time = third_ts.value
@@ -2587,6 +2620,32 @@ def test_batch_restore_version_mixed_as_ofs(lmdb_version_store):
     assert latest["s2"].metadata == "s2-2"
     assert latest["s3"].metadata == "s3-1"
 
+    # check restore from snapshot and negative ver number
+    res = lib.batch_restore_version(syms, [-3, "snap", -1])
+
+    # check returned data
+    assert res[0].symbol == "s1"
+    assert res[1].symbol == "s2"
+    assert res[2].symbol == "s3"
+    assert res[0].version == 3
+    assert res[1].version == 4
+    assert res[2].version == 3  # We restored last version (-1 is last)
+    assert res[0].metadata == "s1-1"
+    assert res[1].metadata == "s2-2"
+    assert res[2].metadata == "s3-1"
+
+    # check latest version of symbols from the read
+    latest = lib.batch_read(syms)
+    assert latest["s1"].version == 3
+    assert latest["s2"].version == 4
+    assert latest["s3"].version == 3
+    assert_equal(latest["s1"].data, first_data[0])
+    assert_equal(latest["s2"].data, second_data[1])
+    assert_equal(latest["s3"].data, first_data[2])
+    assert latest["s1"].metadata == "s1-1"
+    assert latest["s2"].metadata == "s2-2"
+    assert latest["s3"].metadata == "s3-1"
+
 
 @pytest.mark.parametrize("bad_thing", ("symbol", "as_of", "duplicate"))
 def test_batch_restore_version_bad_input_noop(lmdb_version_store, bad_thing):
@@ -2725,6 +2784,15 @@ def test_batch_restore_version(basic_store_tombstone):
         assert_equal(read_df, dfs[d])
 
 
+@pytest.mark.storage
+def test_name_method(basic_store_factory):
+
+    for lib_name in ["my name", "1243"]:
+        lib: NativeVersionStore = basic_store_factory(name=lib_name)
+        assert lib_name == lib.name()
+        lib.version_store.clear()
+
+
 @pytest.mark.storage
 def test_batch_append(basic_store_tombstone, three_col_df):
     lmdb_version_store = basic_store_tombstone