Refactored code with bug fixes and added more tests

Author: Niall Langley

DataPipelineTools.Functions/DataLake/DataLakeConfigFactory.cs

@@ -82,11 +82,11 @@ private IEnumerable<Filter<DataLakeItem>> ParseFilters(HttpRequest req)
             var filters = req.Query.Keys
                             .Where(k => k.StartsWith("filter[") && k.EndsWith("]"))
                             // Clean up the column name by removing the filter[...] parts
-                            .Select(f => f[7..^1])
-                            .SelectMany(k => req.Query[k].Select(v => FilterFactory<DataLakeItem>.Create(k, v, _logger)))
+                            //.Select(f => f[7..^1])
+                            .SelectMany(k => req.Query[k].Select(v => FilterFactory<DataLakeItem>.Create(k[7..^1], v, _logger)))
                             .Where(f => f != null);
 
-            return filters;
+            return filters.ToArray();
         }
 
         private dynamic GetRequestData(HttpRequest req)

DataPipelineTools.Tests/DataLake/DataLakeServiceTests/CheckPathAsyncTests.cs

@@ -179,16 +179,5 @@ public void Given_FilePathWithIncorrectCase_Should_ThrowWhenMultipleFilesMatch()
             var exception = Assert.CatchAsync(() => Sut.CheckPathAsync(testPath, false));
             Assert.That(exception, Is.TypeOf(typeof(Exception)));
         }
-
-
-        //[Test]
-        //public void Given__Should_()
-        //{
-        //    var testPath = "some/invalid/path";
-        //    var resultPath = Sut.CheckPathAsync(testPath, true).Result;
-
-        //    Assert.That(null, Is.EqualTo(resultPath));
-        //}
-
     }
 }

DataPipelineTools.Tests/DataLake/DataLakeServiceTests/GetItemsAsyncTests.cs

@@ -0,0 +1,94 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Moq;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using NUnit.Framework;
+using SqlCollaborative.Azure.DataPipelineTools.DataLake;
+using SqlCollaborative.Azure.DataPipelineTools.DataLake.Model;
+
+namespace DataPipelineTools.Tests.DataLake.DataLakeServiceTests
+{
+    [TestFixture]
+    public class GetItemsAsyncTests : DataLakeTestBase
+    {
+
+        protected readonly DataLakeService Sut;
+        private DataLakeConfig DatalakeConfig => new DataLakeConfig();
+
+        public GetItemsAsyncTests()
+        {
+            // Use the factory to inject the mock logger to get the mock client...
+            var factory = new DataLakeServiceFactory(MockLogger.Object);
+            Sut = factory.CreateDataLakeService(MockFileSystemClient.Object);
+        }
+
+        [SetUp]
+        public void Setup()
+        {
+            // Reset the logger for each test, and add a setup to the moq to write log entries to the console so they are captured
+            // as additional output in the test result
+            MockLogger.Reset();
+            SetupConsoleLogging();
+        }
+
+        [Test]
+        public void Given_ValidDirectoryPath_Should_ReturnContents()
+        {
+            var itemsConfig = new DataLakeGetItemsConfig
+            {
+                Directory = "raw/api/feb"
+            };
+
+            var result = Sut.GetItemsAsync(DatalakeConfig, itemsConfig).Result;
+
+            Assert.That(result.Count, Is.EqualTo(2));
+        }
+
+        [Test]
+        public void Given_DirectoryPathWithIncorrectCase_Should_ReturnContentsForCorrectedPath()
+        {
+            var itemsConfig = new DataLakeGetItemsConfig
+            {
+                Directory = "raw/aPi/feb"
+            };
+
+
+            var r = Sut.CheckPathAsync("raw/Api/jan/delta_extract_1.json", false).Result;
+
+            var result = Sut.GetItemsAsync(DatalakeConfig, itemsConfig).Result;
+
+            Assert.That(result.ContainsKey("fileCount"), Is.True);
+            Assert.That(result.ContainsKey("files"), Is.True);
+
+            Assert.That( (int)result.Property("fileCount").Value, Is.EqualTo(2) );
+
+            var s = result.ToObject<GetItemsResponse>();
+            var responseObject = JsonConvert.DeserializeObject<GetItemsResponse>(result.ToString());
+
+            var files = result.Property("files");
+            var itemsInfo = JsonConvert.DeserializeObject<List<DataLakeItem>>(files.ToString());
+            Assert.That(itemsInfo, Is.EqualTo(2));
+            //Assert.That(itemsInfo.Count(x => x.FullPath == "raw/api/feb"), Is.True);
+            //Assert.That(itemsInfo.Count(x => x.FullPath == "raw/api/feb/delta_extract_3.json"), Is.True);
+        }
+
+        //[Test]
+        //public void Given_ValidDirectoryPath_Should_ReturnDirectoryPath()
+        //{
+        //    var testPath = "raw/database";
+        //    var resultPath = Sut.CheckPathAsync(testPath, true).Result;
+
+        //    Assert.That(resultPath, Is.EqualTo(testPath));
+        //}
+
+    }
+
+    public class GetItemsResponse
+    {
+        public int fileCount { get; set; }
+        public string correctedFilePath { get; set; }
+        public List<DataLakeItem> files { get; set; }
+    }
+}

DataPipelineTools.Tests/DataLake/DataLakeServiceTests/GetItemsAsyncTests_Data_PathItem.csv

@@ -0,0 +1,8 @@
+Name,IsDirectory,ContentLength,LastModified
+raw/database/jan/extract_1.csv,FALSE,10,02/01/2021 13:00
+raw/database/feb/extract_2.csv,FALSE,20,03/01/2021 13:00
+raw/database/feb/EXTRACT_2.csv,FALSE,25,03/01/2021 13:00
+raw/api/jan/delta_extract_1.json,FALSE,10,01/01/2021 14:00
+raw/api/jan/delta_extract_2.json,FALSE,20,02/01/2021 14:00
+raw/api/feb/delta_extract_3.json,FALSE,30,03/01/2021 14:00
+raw/API/jan/delta_extract_1.json,FALSE,10,01/01/2021 15:00

DataPipelineTools.Tests/DataPipelineTools.Tests.csproj

@@ -22,6 +22,9 @@
     <None Update="DataLake\DataLakeServiceTests\CheckPathAsyncTests_Data_PathItem.csv">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="DataLake\DataLakeServiceTests\GetItemsAsyncTests_Data_PathItem.csv">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
 
 </Project>

DataPipelineTools/Common/FilterFactory.cs

@@ -92,6 +92,10 @@ public static Filter<T> Create(string columnName, string filter, ILogger log)
                     DateTime dateTimeVal;
                     isValueParseable = DateTime.TryParse(val, out dateTimeVal);
                     break;
+                case nameof(DateTimeOffset):
+                    DateTimeOffset dateTimeOffsetVal;
+                    isValueParseable = DateTimeOffset.TryParse(val, out dateTimeOffsetVal);
+                    break;
             }
 
             var parseError = isValueParseable ? null : $"The filter '{val}' cannot be applied to the property '{columnName}' as it cannot be cast to a '{propertyType}'";

DataPipelineTools/DataLake/DataLakeService.cs

@@ -47,37 +47,51 @@ public async Task<string> CheckPathAsync(string path, bool isDirectory)
 
             // If the directory does not exist, we find it
             string validDirectory = null;
+            IEnumerable<string> validDirectories = null;
             if (!await _client.GetDirectoryClient(path).ExistsAsync())
             {
-                var directoryParts = directoryPath.Split('/');
-                foreach (var directoryPart in directoryParts)
-                {
-                    var searchItem = directoryPart;
-                    var validPaths = MatchPathItemsCaseInsensitive(validDirectory, searchItem, true);
-
-                    if (validPaths.Count == 0)
-                        return null;
-                    else if (validPaths.Count > 1)
-                        throw new Exception("Multiple paths matched with case insensitive compare.");
-
-                    validDirectory = validPaths[0];
-                }
+                var validPaths = MatchPaths(null, true,directoryPath.Split('/')).ToList();
+                if (validPaths.Count == 0)
+                    return null;
+                if (validPaths.Count > 1 && isDirectory)
+                    throw new Exception("Multiple directories matched with case insensitive compare.");
+
+                validDirectory = validPaths[0];
+                validDirectories = validPaths;
             }
 
+
             if (isDirectory)
                 return validDirectory;
 
             // Now check if the file exists using the corrected directory, and if not find a match...
-            var testFilePath = $"{validDirectory ?? ""}/{filename}".TrimStart('/');
-            if (_client.GetFileClient(testFilePath).Exists())
-                return testFilePath;
+            var files = validDirectories.SelectMany(x => MatchPaths(x, false, filename)).ToList();
 
-            var files = MatchPathItemsCaseInsensitive(validDirectory, filename, false);
             if (files.Count > 1)
-                throw new Exception("Multiple paths matched with case insensitive compare.");
+                throw new Exception("Multiple files matched with case insensitive compare.");
             return files.FirstOrDefault();
         }
 
+        private IEnumerable<string> MatchPaths(string basePath, bool directoriesOnly, params string[] directoryParts)
+        {
+            if (directoryParts == null)
+                return null;
+
+            if (directoryParts.Count() == 0)
+                return new []{ basePath };
+
+            var matchedDirectories = MatchPathItemsCaseInsensitive(basePath, directoryParts.First(), directoriesOnly);
+            var matchedChildDirectories = new List<string>();
+
+            foreach (var directory in matchedDirectories)
+            {
+                var childDirectories = MatchPaths(directory, true, directoryParts.Skip(1).ToArray());
+                matchedChildDirectories.AddRange(childDirectories);
+            }
+
+            return matchedChildDirectories;
+        }
+
         private IList<string> MatchPathItemsCaseInsensitive(string basePath, string searchItem, bool isDirectory)
         {
             var paths = _client.GetPaths(basePath).ToList();
@@ -94,7 +108,7 @@ await CheckPathAsync(getItemsConfig.Directory, true) :
                                 getItemsConfig.Directory;
 
             if (!_client.GetDirectoryClient(directory).Exists())
-                throw new DirectoryNotFoundException("Directory '{directory} could not be found'");
+                throw new DirectoryNotFoundException($"Directory '{directory} could not be found'");
 
             var paths = _client
                 .GetPaths(path: directory ?? string.Empty, recursive: getItemsConfig.Recursive)

DataPipelineTools/DataLake/Model/DataLakeGetItemsConfig.cs

@@ -6,11 +6,11 @@ namespace SqlCollaborative.Azure.DataPipelineTools.DataLake.Model
     public class DataLakeGetItemsConfig
     {
         public string Directory { get; set; }
-        public bool IgnoreDirectoryCase { get; set; }
-        public bool Recursive { get; set; }
+        public bool IgnoreDirectoryCase { get; set; } = true;
+        public bool Recursive { get; set; } = true;
         public string OrderByColumn { get; set; }
         public bool OrderByDescending { get; set; }
-        public int Limit { get; set; }
-        public IEnumerable<Filter<DataLakeItem>> Filters { get; set; }
+        public int Limit { get; set; } = 0;
+        public IEnumerable<Filter<DataLakeItem>> Filters { get; set; } = new Filter<DataLakeItem>[0];
     }
 }