Repository: ludwig-ai/ludwig
Pull request: fix: Updated fs_utils to get valid windows path from url
Resolves #1802 which was having an issue writing to a windows file path after calling `get_fs_and_path`

```
File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\utils\fs_utils.py:133, in open_file(url, *args, **kwargs)
    [130](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=129) @contextlib.contextmanager
    [131](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=130) def open_file(url, *args, **kwargs):
    [132](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=131)     fs, path = get_fs_and_path(url)
--> [133](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=132)     with fs.open(path, *args, **kwargs) as f:
    [134](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=133)         yield f
```

The `fs_utils.get_fs_and_path` method that calls `fsspec.split_protocol` which does to support conversion to windows paths [windows paths](https://github.com/fsspec/filesystem_spec/blob/master/fsspec/core.py#L518).

Updated this method to parse the url path, and decode any url escaping before using [pathlib.PurePosixPath](https://docs.python.org/3/library/pathlib.html) to convert the url into relative os path.

Problem statement:
Problem logging an experiment to wandb
**Describe the bug**
Logging a training run works fine, however when logging an experiment run (the evaluation part) with the default parameters, there is a OSError 22 relating to "probabilities_<UNK>.csv"

`OSError: [Errno 22] Invalid argument: '/results/classification_Distilbert_2/label_probabilities_<UNK>.csv`

**To Reproduce**
Steps to reproduce the behavior:

1. Run an experiment from the CLI with default parameters.
2. Configure the model for text classification.
3. Add the wandb flag

**Model config**

```
preprocessing:
    text:
        word_tokenizer: hf_tokenizer

input_features:
    -
        name: feature1
        type: text
        level: word
        encoder: distilbert
        pretrained_model_name_or_path: distilbert-base-uncased
    -
        name: feature2
        type: text
        level: word
        encoder: distilbert
        pretrained_model_name_or_path: distilbert-base-uncased
        tied_weights: feature1
    -
        name: feature3
        type: category

output_features:
    -
        name: label
        top_k: 5
        type: category

training:
    epochs: 3
    batch_size: 16
    learning_rate: 0.000005
```

**wandb metadata**

```
{
    "os": "Windows-10-10.0.19041-SP0",
    "python": "3.9.0",
    "heartbeatAt": "2022-03-10T14:21:23.463980",
    "startedAt": "2022-03-10T14:21:17.632459",
    "docker": null,
    "gpu": "NVIDIA GeForce RTX 2070",
    "gpu_count": 1,
    "cpu_count": 6,
    "cuda": null,
    "args": [
        "experiment",
        "--output_directory",
        "results",
        "--experiment_name",
        "classification",
        "--model_name",
        "Distilbert",
        "--training_set",
        "TRAIN_en.arrow",
        "--validation_set",
        "VAL_en.arrow",
        "--test_set",
        "EVAL_en.arrow",
        "--data_format",
        "feather",
        "--config_file",
        "model1.yaml",
        "--wandb"
    ],
    "state": "running",
    "program": "-m __main__",
    "executable": "I:\\Anaconda\\envs\\ludwig\\python.exe"
}
```

**Expected behavior**
The csv file for the UNK token should be created.


**Environment**

- OS: Windows-10-10.0.19041-SP0
- Python version: 3.9
- Ludwig version: 0.4.1

**Additional context**
The problem can be avoided with the ludwig API and passing the parameter skip_save_predictions=True (not possible with the CLI).

Hints from issue discussion:
Hi @danielduckworth thanks for raising this issue, on the surface this issues looks to be related to the difference between windows and unix file paths.  Are you able to share a small sample of the dataset so that I might try and reproduce in a windows environment?
Unfortunately not the original data I was using. But I'll see if I can reproduce the problem with public data and send that.
@danielduckworth what about using [Ludwig's `synthesize_dataset` command](https://ludwig-ai.github.io/ludwig-docs/0.4/user_guide/command_line_interface/#synthesize_dataset)?
Also @danielduckworth could you provide more context on the error, like the full stack trace by any chance? The reason I'm asking is that looking at the code it seems we are always creating paths with the os libraries, so running on windows you should not see those linux like paths at all, I'm curious in which part of the code they are popping up.
Hi @w4nderlust I was able to reproduce the error using the synthesizer function (and without wandb logging).

Jupyter notebook here: [https://github.com/danielduckworth/ludwig-os-error](https://github.com/danielduckworth/ludwig-os-error)

The repo also includes an export of the conda environment as yaml.

I'm using VS Code with the Jupyter extension: https://marketplace.visualstudio.com/items?itemName=ms-toolsai.jupyter

**Full stack trace:**



```
---------------------------------------------------------------------------
OSError                                   Traceback (most recent call last)
Input In [3], in <cell line: 6>()
      2     config = yaml.safe_load(f.read())
      4 model = LudwigModel(config=config)
----> 6 model = model.experiment(
      7     dataset=dataset_file_path,
      8     data_format="csv",
      9     experiment_name="test",
     10     model_name="model1",
     11     model_resume_path=None, 
     12     skip_save_log=False, 
     13     skip_save_unprocessed_output=True,
     14 )

File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\api.py:1159, in LudwigModel.experiment(self, dataset, training_set, validation_set, test_set, training_set_metadata, data_format, experiment_name, model_name, model_load_path, model_resume_path, eval_split, skip_save_training_description, skip_save_training_statistics, skip_save_model, skip_save_progress, skip_save_log, skip_save_processed_input, skip_save_unprocessed_output, skip_save_predictions, skip_save_eval_stats, skip_collect_predictions, skip_collect_overall_stats, output_directory, random_seed, debug, **kwargs)
   [1157](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1156) # predict
   [1158](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1157) try:
-> [1159](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1158)     eval_stats, _, _ = self.evaluate(
   [1160](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1159)         eval_set,
   [1161](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1160)         data_format=data_format,
   [1162](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1161)         batch_size=batch_size,
   [1163](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1162)         output_directory=output_directory,
   [1164](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1163)         skip_save_unprocessed_output=skip_save_unprocessed_output,
   [1165](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1164)         skip_save_predictions=skip_save_predictions,
   [1166](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1165)         skip_save_eval_stats=skip_save_eval_stats,
   [1167](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1166)         collect_predictions=not skip_collect_predictions,
   [1168](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1167)         collect_overall_stats=not skip_collect_overall_stats,
   [1169](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1168)         return_type='dict',
   [1170](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1169)         debug=debug
   [1171](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1170)     )
   [1172](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1171) except NotImplementedError:
   [1173](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1172)     logger.warning(
   [1174](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1173)         "Skipping evaluation as the necessary methods are not "
   [1175](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1174)         "supported. Full exception below:\n"
   [1176](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1175)         f"{traceback.format_exc()}"
   [1177](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=1176)     )

File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\api.py:950, in LudwigModel.evaluate(self, dataset, data_format, split, batch_size, skip_save_unprocessed_output, skip_save_predictions, skip_save_eval_stats, collect_predictions, collect_overall_stats, output_directory, return_type, debug, **kwargs)
    [944](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=943) should_save_predictions = (
    [945](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=944)     collect_predictions
    [946](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=945)     and postproc_predictions is not None
    [947](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=946)     and not skip_save_predictions
    [948](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=947) )
    [949](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=948) if should_save_predictions:
--> [950](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=949)     save_prediction_outputs(
    [951](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=950)         postproc_predictions, 
    [952](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=951)         self.model.output_features, 
    [953](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=952)         output_directory, 
    [954](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=953)         self.backend
    [955](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=954)     )
    [957](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=956) print_evaluation_stats(eval_stats)
    [958](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/api.py?line=957) if not skip_save_eval_stats:

File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\models\predictor.py:373, in save_prediction_outputs(postprocessed_output, output_features, output_directory, backend)
    [371](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/models/predictor.py?line=370) for output_field, outputs in postprocessed_dict.items():
    [372](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/models/predictor.py?line=371)     for output_type, values in outputs.items():
--> [373](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/models/predictor.py?line=372)         save_csv(csv_filename.format(output_field, output_type), values)

File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\utils\data_utils.py:201, in save_csv(data_fp, data)
    [200](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/data_utils.py?line=199) def save_csv(data_fp, data):
--> [201](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/data_utils.py?line=200)     with open_file(data_fp, 'w', encoding='utf-8') as csv_file:
    [202](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/data_utils.py?line=201)         writer = csv.writer(csv_file)
    [203](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/data_utils.py?line=202)         for row in data:

File I:\Anaconda\envs\ludwig5\lib\contextlib.py:119, in _GeneratorContextManager.__enter__(self)
    [117](file:///i%3A/Anaconda/envs/ludwig5/lib/contextlib.py?line=116) del self.args, self.kwds, self.func
    [118](file:///i%3A/Anaconda/envs/ludwig5/lib/contextlib.py?line=117) try:
--> [119](file:///i%3A/Anaconda/envs/ludwig5/lib/contextlib.py?line=118)     return next(self.gen)
    [120](file:///i%3A/Anaconda/envs/ludwig5/lib/contextlib.py?line=119) except StopIteration:
    [121](file:///i%3A/Anaconda/envs/ludwig5/lib/contextlib.py?line=120)     raise RuntimeError("generator didn't yield") from None

File I:\Anaconda\envs\ludwig5\lib\site-packages\ludwig\utils\fs_utils.py:133, in open_file(url, *args, **kwargs)
    [130](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=129) @contextlib.contextmanager
    [131](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=130) def open_file(url, *args, **kwargs):
    [132](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=131)     fs, path = get_fs_and_path(url)
--> [133](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=132)     with fs.open(path, *args, **kwargs) as f:
    [134](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/ludwig/utils/fs_utils.py?line=133)         yield f

File I:\Anaconda\envs\ludwig5\lib\site-packages\fsspec\spec.py:997, in AbstractFileSystem.open(self, path, mode, block_size, cache_options, compression, **kwargs)
    [989](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=988)     mode = mode.replace("t", "") + "b"
    [991](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=990)     text_kwargs = {
    [992](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=991)         k: kwargs.pop(k)
    [993](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=992)         for k in ["encoding", "errors", "newline"]
    [994](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=993)         if k in kwargs
    [995](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=994)     }
    [996](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=995)     return io.TextIOWrapper(
--> [997](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=996)         self.open(
    [998](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=997)             path,
    [999](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=998)             mode,
   [1000](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=999)             block_size=block_size,
   [1001](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1000)             cache_options=cache_options,
   [1002](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1001)             compression=compression,
   [1003](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1002)             **kwargs,
   [1004](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1003)         ),
   [1005](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1004)         **text_kwargs,
   [1006](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1005)     )
   [1007](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1006) else:
   [1008](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1007)     ac = kwargs.pop("autocommit", not self._intrans)

File I:\Anaconda\envs\ludwig5\lib\site-packages\fsspec\spec.py:1009, in AbstractFileSystem.open(self, path, mode, block_size, cache_options, compression, **kwargs)
   [1007](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1006) else:
   [1008](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1007)     ac = kwargs.pop("autocommit", not self._intrans)
-> [1009](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1008)     f = self._open(
   [1010](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1009)         path,
   [1011](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1010)         mode=mode,
   [1012](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1011)         block_size=block_size,
   [1013](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1012)         autocommit=ac,
   [1014](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1013)         cache_options=cache_options,
   [1015](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1014)         **kwargs,
   [1016](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1015)     )
   [1017](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1016)     if compression is not None:
   [1018](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/spec.py?line=1017)         from fsspec.compression import compr

File I:\Anaconda\envs\ludwig5\lib\site-packages\fsspec\implementations\local.py:155, in LocalFileSystem._open(self, path, mode, block_size, **kwargs)
    [153](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=152) if self.auto_mkdir and "w" in mode:
    [154](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=153)     self.makedirs(self._parent(path), exist_ok=True)
--> [155](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=154) return LocalFileOpener(path, mode, fs=self, **kwargs)

File I:\Anaconda\envs\ludwig5\lib\site-packages\fsspec\implementations\local.py:250, in LocalFileOpener.__init__(self, path, mode, autocommit, fs, compression, **kwargs)
    [248](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=247) self.compression = get_compression(path, compression)
    [249](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=248) self.blocksize = io.DEFAULT_BUFFER_SIZE
--> [250](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=249) self._open()

File I:\Anaconda\envs\ludwig5\lib\site-packages\fsspec\implementations\local.py:255, in LocalFileOpener._open(self)
    [253](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=252) if self.f is None or self.f.closed:
    [254](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=253)     if self.autocommit or "w" not in self.mode:
--> [255](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=254)         self.f = open(self.path, mode=self.mode)
    [256](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=255)         if self.compression:
    [257](file:///i%3A/Anaconda/envs/ludwig5/lib/site-packages/fsspec/implementations/local.py?line=256)             compress = compr[self.compression]

OSError: [Errno 22] Invalid argument: 'i:/Google Drive/Programming/03 Projects/ISCO/pytorch_train/results/test_model1_2/label_probabilities_<UNK>.csv
```
Thanks for this detailed error log @danielduckworth.  We have identified the issue as a result of converting from a url which has the `Posix` file system format, to a local windows path.  I have PR #1830 which should resolve this issue.

Please note whilst best effort is made to work on windows, it is not an officially supported platform at the moment.
Thanks @brightsparc. I'll look into setting up a dev environment with wsl for ludwig.

Commit message: Refactor assert handling for fs, and dded pytest filesystem marker

Patch to review:
diff --git a/ludwig/utils/fs_utils.py b/ludwig/utils/fs_utils.py
index 26555a79f54..5f83d312615 100644
--- a/ludwig/utils/fs_utils.py
+++ b/ludwig/utils/fs_utils.py
@@ -18,6 +18,7 @@
 import os
 import pathlib
 import tempfile
+from urllib.parse import unquote, urlparse
 
 import fsspec
 import h5py
@@ -27,6 +28,10 @@
 
 def get_fs_and_path(url):
     protocol, path = split_protocol(url)
+    # Parse the url to get only the escaped url path
+    path = unquote(urlparse(path).path)
+    # Create a windows compatible path from url path
+    path = os.fspath(pathlib.PurePosixPath(path))
     fs = fsspec.filesystem(protocol)
     return fs, path
 
diff --git a/pytest.ini b/pytest.ini
index f64797d5da0..f4009b8287f 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,4 +1,5 @@
 [pytest]
 markers =
     distributed: mark a test as a distributed test.
+    filesystem: mark to test operating system systems.
     slow: mark test as slow.
diff --git a/tests/ludwig/utils/test_fs_utils.py b/tests/ludwig/utils/test_fs_utils.py
new file mode 100644
index 00000000000..0268fe1724e
--- /dev/null
+++ b/tests/ludwig/utils/test_fs_utils.py
@@ -0,0 +1,87 @@
+# Copyright (c) 2019 Uber Technologies, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+import logging
+import os
+import platform
+import tempfile
+from urllib.parse import quote
+
+import pytest
+
+from ludwig.utils.fs_utils import get_fs_and_path
+
+
+def create_file(url):
+    _, path = get_fs_and_path(url)
+    logging.info(f"saving url '{url}' to path '{path}'")
+    with tempfile.TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, path)
+        os.makedirs(os.path.dirname(file_path))
+        with open(file_path, "w"):
+            return path
+
+
+@pytest.mark.filesystem
+def test_get_fs_and_path_simple():
+    assert create_file("http://a/b.jpg") == os.path.join("a", "b.jpg")
+
+
+@pytest.mark.filesystem
+def test_get_fs_and_path_query_string():
+    assert create_file("http://a/b.jpg?c=d") == os.path.join("a", "b.jpg")
+
+
+@pytest.mark.filesystem
+def test_get_fs_and_path_decode():
+    assert create_file("http://a//b%20c.jpg") == os.path.join("a", "b c.jpg")
+
+
+@pytest.mark.filesystem
+def test_get_fs_and_path_unicode():
+    assert create_file("http://a/æ.jpg") == "a/æ.jpg"
+
+
+@pytest.mark.filesystem
+@pytest.mark.skipif(platform.system() == "Windows", reason="Skipping if windows.")
+def test_get_fs_and_path_invalid_linux():
+    invalid_chars = {
+        "\x00": ValueError,
+        "/": FileExistsError,
+    }
+    for c, e in invalid_chars.items():
+        url = f"http://a/{quote(c)}"
+        with pytest.raises(e):
+            create_file(url)
+
+
+@pytest.mark.filesystem
+@pytest.mark.skipif(platform.system() != "Windows", reason="Skipping if not windows.")
+def test_get_fs_and_path_invalid_windows():
+    invalid_chars = {
+        "\x00": ValueError,
+        "\\": FileExistsError,
+        "/": OSError,
+        ":": OSError,
+        "*": OSError,
+        "?": OSError,
+        '"': OSError,
+        "<": OSError,
+        ">": OSError,
+        "|": OSError,
+    }
+    for c, e in invalid_chars.items():
+        url = f"http://a/{quote(c)}"
+        with pytest.raises(e):
+            create_file(url)


Code review comments:
diff_hunk: @@ -0,0 +1,87 @@
+# Copyright (c) 2019 Uber Technologies, Inc.
path: tests/ludwig/utils/test_fs_utils.py
text: @user1:
nit: Remove the header for new files.