fixing import/export tests

Author: nucleogenesis

kolibri/core/auth/management/commands/bulkexportusers.py

@@ -246,8 +246,8 @@ def get_facility(self, options):
 
     def get_filename(self, options, facility):
         if options["output_file"] is None:
-            filename = CSV_EXPORT_FILENAMES["user"].format(
-                facility.name, facility.id[:4]
+            filename = default_storage.get_available_name(
+                CSV_EXPORT_FILENAMES["user"].format(facility.name, facility.id[:4])
             )
         else:
             filename = options["output_file"]

kolibri/core/auth/management/commands/importusers.py

@@ -197,8 +197,7 @@ def handle(self, *args, **options):
 
         fieldnames = input_fields + tuple(val for val in labels.values())
 
-        csv_file = open_csv_for_reading(options["filepath"])
-        with csv_file as f:
+        with open_csv_for_reading(options["filepath"]) as f:
             header = next(csv.reader(f, strict=True))
             has_header = False
             if all(col in fieldnames for col in header):
@@ -213,8 +212,7 @@ def handle(self, *args, **options):
                     "Mix of valid and invalid header labels found in first row"
                 )
 
-        csv_file = open_csv_for_reading(options["filepath"])
-        with csv_file as f:
+        with open_csv_for_reading(options["filepath"]) as f:
             if has_header:
                 reader = csv.DictReader(f, strict=True)
             else:

kolibri/core/auth/test/test_user_import.py

@@ -1,7 +1,6 @@
 import csv
-import os
-import tempfile
 
+from django.core.files.storage import default_storage
 from django.core.management import call_command
 from django.core.management.base import CommandError
 from django.test import TestCase
@@ -137,11 +136,8 @@ def test_create_user_not_exist_bad_username(self):
 
 class DeviceNotSetup(TestCase):
     def test_device_not_setup(self):
-        csvfile, csvpath = tempfile.mkstemp(suffix="csv")
         with self.assertRaisesRegex(CommandError, "No default facility exists"):
-            call_command("importusers", csvpath)
-        os.close(csvfile)
-        os.remove(csvpath)
+            call_command("importusers", "you_cannot_import_yet.csv")
 
 
 class UserImportCommandTestCase(TestCase):
@@ -157,34 +153,32 @@ def setUpClass(self):
         self.facility, self.superuser = setup_device()
 
     def setUp(self):
-        self.csvfile, self.csvpath = tempfile.mkstemp(suffix="csv")
+        self.csvfilename = default_storage.get_available_name("test.csv")
 
     def tearDown(self):
         FacilityUser.objects.exclude(username=self.superuser.username).delete()
-        os.close(self.csvfile)
-        os.remove(self.csvpath)
+        default_storage.delete(self.csvfilename)
 
     def importFromRows(self, *args):
-        csv_file = open_csv_for_writing(self.csvpath)
-        with csv_file as f:
+        with open_csv_for_writing(self.csvfilename) as f:
             writer = csv.writer(f)
             writer.writerows([a for a in args])
 
-        call_command("importusers", self.csvpath)
+        call_command("importusers", self.csvfilename)
 
     def test_setup_headers_no_username(self):
         with self.assertRaisesRegex(CommandError, "No usernames specified"):
             self.importFromRows(["class", "facility"])
-            call_command("importusers", self.csvpath)
+            call_command("importusers", self.csvfilename)
 
     def test_setup_headers_invalid_header(self):
         with self.assertRaisesRegex(CommandError, "Mix of valid and invalid header"):
             self.importFromRows(["class", "facility", "dogfood"])
-            call_command("importusers", self.csvpath)
+            call_command("importusers", self.csvfilename)
 
     def test_setup_headers_make_user(self):
         self.importFromRows(["username"], ["testuser"])
-        call_command("importusers", self.csvpath)
+        call_command("importusers", self.csvfilename)
         self.assertTrue(FacilityUser.objects.filter(username="testuser").exists())
 
     def test_setup_no_headers_make_user(self):
@@ -271,10 +265,13 @@ def test_import_from_export_csv(self):
         for user in users:
             FacilityUser.objects.create(facility=self.facility, **user)
         call_command(
-            "exportusers", output_file=self.csvpath, overwrite=True, demographic=True
+            "exportusers",
+            output_file=self.csvfilename,
+            overwrite=True,
+            demographic=True,
         )
         FacilityUser.objects.all().delete()
-        call_command("importusers", self.csvpath)
+        call_command("importusers", self.csvfilename)
         for user in users:
             user_model = FacilityUser.objects.get(username=user["username"])
             self.assertEqual(user_model.gender, user["gender"])
@@ -284,16 +281,18 @@ def test_import_from_export_csv(self):
     def test_import_from_export_missing_headers(self):
         for user in users:
             FacilityUser.objects.create(facility=self.facility, **user)
+
         call_command(
-            "exportusers", output_file=self.csvpath, overwrite=True, demographic=True
+            "exportusers",
+            output_file=self.csvfilename,
+            overwrite=True,
+            demographic=True,
         )
         cols_to_remove = ["Facility id", "Gender"]
-        csv_file = open_csv_for_reading(self.csvpath)
-        with csv_file as source:
+        with open_csv_for_reading(self.csvfilename) as source:
             reader = csv.DictReader(source)
             rows = [row for row in reader]
-        csv_file = open_csv_for_writing(self.csvpath)
-        with csv_file as result:
+        with open_csv_for_writing("new" + self.csvfilename) as result:
             writer = csv.DictWriter(
                 result,
                 tuple(
@@ -305,8 +304,9 @@ def test_import_from_export_missing_headers(self):
                 for col in cols_to_remove:
                     del row[col]
                 writer.writerow(row)
+
         FacilityUser.objects.all().delete()
-        call_command("importusers", self.csvpath)
+        call_command("importusers", "new" + self.csvfilename)
         for user in users:
             user_model = FacilityUser.objects.get(username=user["username"])
             self.assertEqual(user_model.birth_year, user["birth_year"])
@@ -316,15 +316,16 @@ def test_import_from_export_mixed_headers(self):
         for user in users:
             FacilityUser.objects.create(facility=self.facility, **user)
         call_command(
-            "exportusers", output_file=self.csvpath, overwrite=True, demographic=True
+            "exportusers",
+            output_file=self.csvfilename,
+            overwrite=True,
+            demographic=True,
         )
         cols_to_replace = {"Facility id": "facility", "Gender": "gender"}
-        csv_file = open_csv_for_reading(self.csvpath)
-        with csv_file as source:
+        with open_csv_for_reading(self.csvfilename) as source:
             reader = csv.DictReader(source)
             rows = [row for row in reader]
-        csv_file = open_csv_for_writing(self.csvpath)
-        with csv_file as result:
+        with open_csv_for_writing("new" + self.csvfilename) as result:
             writer = csv.DictWriter(
                 result,
                 tuple(
@@ -339,7 +340,7 @@ def test_import_from_export_mixed_headers(self):
                     del row[col]
                 writer.writerow(row)
         FacilityUser.objects.all().delete()
-        call_command("importusers", self.csvpath)
+        call_command("importusers", "new" + self.csvfilename)
         for user in users:
             user_model = FacilityUser.objects.get(username=user["username"])
             self.assertEqual(user_model.birth_year, user["birth_year"])

kolibri/core/logger/csv_export.py

@@ -2,7 +2,6 @@
 import datetime
 import logging
 import math
-import os
 from collections import OrderedDict
 
 from dateutil import parser
@@ -11,6 +10,7 @@
 from django.db.models import Max
 from django.db.models import OuterRef
 from django.db.models import Subquery
+from django.core.files.storage import default_storage
 from django.utils.translation import gettext_lazy as _
 from django.utils.translation import pgettext_lazy
 from le_utils.constants import content_kinds
@@ -256,7 +256,7 @@ def csv_file_generator(
         else parser.parse(end_date) + datetime.timedelta(days=1)
     )
 
-    if not overwrite and os.path.exists(filepath):
+    if not overwrite and default_storage.exists(filepath):
         raise ValueError("{} already exists".format(filepath))
     queryset = log_info["queryset"].filter(
         dataset_id=facility.dataset_id,
@@ -285,9 +285,7 @@ def csv_file_generator(
         label for _, label in topic_headers
     ]
 
-    csv_file = open_csv_for_writing(filepath)
-
-    with csv_file as f:
+    with open_csv_for_writing(filepath) as f:
         writer = csv.DictWriter(f, header_labels)
         logger.info("Creating csv file {filename}".format(filename=filepath))
         writer.writeheader()

kolibri/core/logger/tasks.py

@@ -1,5 +1,6 @@
 import os
 
+from django.core.files.storage import default_storage
 from django.core.management import call_command
 from rest_framework import serializers
 
@@ -12,18 +13,13 @@
 from kolibri.core.tasks.decorators import register_task
 from kolibri.core.tasks.permissions import IsAdminForJob
 from kolibri.core.tasks.validation import JobValidator
-from kolibri.utils import conf
 
 LOGS_CLEANUP_JOB_ID = "18"
 
 
 def get_filepath(log_type, facility_id, start_date, end_date):
     facility = Facility.objects.get(id=facility_id)
-    logs_dir = os.path.join(conf.KOLIBRI_HOME, "log_export")
-    if not os.path.isdir(logs_dir):
-        os.mkdir(logs_dir)
-    filepath = os.path.join(
-        logs_dir,
+    filepath = default_storage.path(
         CSV_EXPORT_FILENAMES[log_type].format(
             facility.name, facility.id[:4], start_date[:10], end_date[:10]
         ),
@@ -177,10 +173,8 @@ def log_exports_cleanup():
     Cleanup log_exports csv files that does not have
     related reocord in GenerateCSVLogRequest model
     """
-    logs_dir = os.path.join(conf.KOLIBRI_HOME, "log_export")
-    if not os.path.isdir(logs_dir):
-        return
     valid_filenames_set = get_valid_filenames()
-    for filename in os.listdir(logs_dir):
+    _, files_in_storage = default_storage.listdir("")
+    for filename in files_in_storage:
         if filename not in valid_filenames_set:
-            os.remove(os.path.join(logs_dir, filename))
+            default_storage.delete(filename)