test_basic.py

#-*- coding: utf-8 -*-
# #############################################################################
# Copyright 2018 Hoffmann-La Roche
#
# 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.
# #############################################################################

from datetime import datetime, timedelta, date
import unittest
import os
import sys
import shutil

import pandas as pd
import numpy as np

is_pathlib_available = False
try:
    from pathlib import Path
    is_pathlib_available = True
except:
    pass


class TestBasic(unittest.TestCase):
    """
    Test suite for pyreadstat.
    """

    def _prepare_data(self):

        self.script_folder = os.path.dirname(os.path.realpath(__file__))
        self.parent_folder = os.path.split(self.script_folder)[0]
        self.data_folder = os.path.join(self.parent_folder, "test_data")
        self.basic_data_folder = os.path.join(self.data_folder, "basic")
        self.catalog_data_folder = os.path.join(self.data_folder, "sas_catalog")
        self.international_data_folder = os.path.join(self.data_folder, "ínternátionál")
        self.missing_data_folder = os.path.join(self.data_folder, "missing_data")
        self.mr_data_folder = os.path.join(self.data_folder, "multiple_response")
        self.write_folder = os.path.join(self.data_folder, "write")
        if not os.path.isdir(self.write_folder):
            os.makedirs(self.write_folder)


        # basic
        pandas_csv = os.path.join(self.basic_data_folder, "sample.csv")
        df_pandas = pd.read_csv(pandas_csv)
        df_pandas["mydate"] = [datetime.strptime(x, '%Y-%m-%d').date() if type(x) == str else float('nan') for x in df_pandas["mydate"]]
        df_pandas["dtime"] = [datetime.strptime(x, '%Y-%m-%dT%H:%M:%S.000000') if type(x) == str else float('nan') for x in
                              df_pandas["dtime"]]
        df_pandas["mytime"] = [datetime.strptime(x, '%H:%M:%S.000000').time() if type(x) == str else float('nan') for x in
                               df_pandas["mytime"]]
        df_pandas["myord"] = df_pandas["myord"].astype(float)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(float)
        self.df_pandas = df_pandas
        # formatted
        mylabl_format = {1.0:"Male", 2.0:"Female"}
        myord_format = {1.0:"low", 2.0:"medium", 3.0:"high"}
        df_pandas_formatted = df_pandas.copy()
        df_pandas_formatted["mylabl"] = df_pandas_formatted["mylabl"].apply(lambda x: mylabl_format[x])
        df_pandas_formatted["myord"] = df_pandas_formatted["myord"].apply(lambda x: myord_format[x])
        df_pandas_formatted["mylabl"] = df_pandas_formatted["mylabl"].astype("category")
        df_pandas_formatted["myord"] = df_pandas_formatted["myord"].astype("category")
        self.df_pandas_formatted = df_pandas_formatted
        # skip some columns
        self.usecols = ['mynum', 'myord']
        cols_to_drop = list(set(df_pandas.columns.values.tolist()) - set(self.usecols))
        self.df_usecols = df_pandas.drop(cols_to_drop, axis=1)
        # sas formatted
        sas_formatted = os.path.join(self.catalog_data_folder, "sas_formatted.csv")
        df_sas = pd.read_csv(sas_formatted)
        df_sas["SEXA"] = df_sas["SEXA"].astype("category")
        df_sas["SEXB"] = df_sas["SEXB"].astype("category")
        self.df_sas_format = df_sas
        # dates
        sas_dates = os.path.join(self.basic_data_folder, "dates.csv")
        df_dates1 = pd.read_csv(sas_dates)
        df_dates1["date"] = pd.to_datetime(df_dates1["date"])
        df_dates1["dtime"] = pd.to_datetime(df_dates1["dtime"])
        df_dates1["time"] = pd.to_datetime(df_dates1["time"], format='%H:%M:%S')
        df_dates1["time"] = df_dates1["time"].apply(lambda x: x.time())
        self.df_sas_dates_as_pandas = df_dates1

        df_dates2 = df_dates1.copy()
        df_dates2["date"] = df_dates2["date"].apply(lambda x: x.date())
        self.df_sas_dates = df_dates2
        self.df_sas_dates2 = pd.concat([self.df_sas_dates, pd.DataFrame([[np.nan, pd.NaT, np.nan]],columns=["date", "dtime", "time"])], ignore_index=True)

        # missing data
        pandas_missing_sav_csv = os.path.join(self.basic_data_folder, "sample_missing.csv")
        df_missing_sav = pd.read_csv(pandas_missing_sav_csv, na_values="#NULL!", keep_default_na=False)
        df_missing_sav["mydate"] = [datetime.strptime(x, '%Y-%m-%d').date() if type(x) == str else float('nan') for x in
                                    df_missing_sav["mydate"]]
        df_missing_sav["dtime"] = [datetime.strptime(x, '%Y-%m-%dT%H:%M:%S.000000') if type(x) == str else float('nan') for x
                              in df_missing_sav["dtime"]]
        df_missing_sav["mytime"] = [datetime.strptime(x, '%H:%M:%S.000000').time() if type(x) == str else float('nan') for x
                               in df_missing_sav["mytime"]]
        self.df_missing_sav = df_missing_sav

        pandas_missing_user_sav_csv = os.path.join(self.basic_data_folder, "sample_missing_user.csv")
        df_user_missing_sav = pd.read_csv(pandas_missing_user_sav_csv, na_values="#NULL!", keep_default_na=False)
        df_user_missing_sav["mydate"] = [datetime.strptime(x, '%Y-%m-%d').date() if type(x) == str else float('nan') for x in
                                         df_user_missing_sav["mydate"]]
        df_user_missing_sav["dtime"] = [datetime.strptime(x, '%Y-%m-%dT%H:%M:%S.000000') if type(x) == str else float('nan')
                                   for x in df_user_missing_sav["dtime"]]
        df_user_missing_sav["mytime"] = [datetime.strptime(x, '%H:%M:%S.000000').time() if type(x) == str else float('nan')
                                    for x in df_user_missing_sav["mytime"]]
        self.df_user_missing_sav = df_user_missing_sav

        # no dates
        nodates_spss_csv = os.path.join(self.basic_data_folder, "sample_nodate_spss.csv")
        df_nodates_spss = pd.read_csv(nodates_spss_csv)
        df_nodates_spss["myord"] = df_nodates_spss["myord"].astype(float)
        df_nodates_spss["mylabl"] = df_nodates_spss["mylabl"].astype(float)
        self.df_nodates_spss = df_nodates_spss

        nodates_sastata_csv = os.path.join(self.basic_data_folder, "sample_nodate_sas_stata.csv")
        df_nodates_sastata = pd.read_csv(nodates_sastata_csv)
        df_nodates_sastata["myord"] = df_nodates_sastata["myord"].astype(float)
        df_nodates_sastata["mylabl"] = df_nodates_sastata["mylabl"].astype(float)
        self.df_nodates_sastata = df_nodates_sastata

        # character column with nan and object column with nan (object pyreadstat writer doesn't know what to do with)
        self.df_charnan = pd.DataFrame([[0,np.nan,np.nan],[1,"test", timedelta]], columns = ["integer", "string", "object"])

        # xport files v5 vs v8
        self.xptv5v8 = pd.DataFrame([[float(x)] for x in range(1,11)], columns=["i"])

        # long string

        self.df_longstr = pd.DataFrame({
                "v1": {
                    "10001": """Lorem ipsum dolor sit amet, consectetur adipiscing elit. Maecenas ac pretium sem. Fusce aliquet
                    augue rhoncus consequat pulvinar. In est ex, porta congue diam sed, laoreet suscipit purus. Phasellus mollis
                    lobortis tellus at vehicula. Etiam egestas augue id massa bibendum volutpat id et ipsum. Praesent ut lorem
                    rhoncus, pharetra risus sed, pharetra sem. In pulvinar egestas erat, id condimentum tortor tempus sed. Duis
                    ornare lacus ut ligula congue, non convallis urna dignissim. Etiam vehicula turpis sit amet nisi finibus
                    laoreet. Duis molestie consequat nulla, non lobortis est tempus sit amet. Quisque elit est,
                    congue non commodo vitae, porttitor ac erat. """,
                    "10002": "fgsdghshsgh",
                    "10003": "gsfdgsdg",
                },
                "v2": {
                    "10001": "gsfdgsfdgsfg",
                    "10002": "fgsdghshsgh",
                    "10003": "gsfdgsdg",
                },

            })

    def setUp(self):

        # set paths
        self._prepare_data()

    def test_sas7bdat(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"))
        self.assertTrue(df.equals(self.df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))
        #self.assertTrue(meta.creation_time==datetime(2018, 8, 16, 18, 21, 52))
        #self.assertTrue(meta.modification_time==datetime(2018, 8, 16, 18, 21, 52))

    def test_sas7bdat_bincompressed(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample_bincompressed.sas7bdat"))
        self.assertTrue(df.equals(self.df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))

    def test_sas7bdat_metaonly(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"))
        df2, meta2 = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta.number_rows == meta2.number_rows)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)
        self.assertTrue(meta.readstat_variable_types["mychar"]=="string")
        self.assertTrue(meta.readstat_variable_types["myord"]=="double")
        self.assertTrue(meta.readstat_variable_types["dtime"]=="double")

    def test_sas7bdat_usecols(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"), usecols=self.usecols)
        self.assertTrue(df.equals(self.df_usecols))
        self.assertTrue(meta.number_columns == len(self.usecols))
        self.assertTrue(meta.column_names == self.usecols)
        
    def test_sas7bdat_international(self):
        """
        On windows, paths with international characters are problematic. This is verifying that it is working as expected
        """
        # in addition, this works only in python 3
        if sys.version_info[0]>2:
            df, meta = pyreadstat.read_sas7bdat(os.path.join(self.international_data_folder, "sample.sas7bdat"))
            self.assertTrue(df.equals(self.df_pandas))
            self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
            self.assertTrue(meta.number_rows == len(self.df_pandas))

    def test_sas7bdat_nodates(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"), disable_datetime_conversion=True)
        self.assertTrue(df.equals(self.df_nodates_sastata))

    def test_sas7bdat_chunk(self):
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "sample.sas7bdat"), row_limit = 2, row_offset =1)
        df_pandas = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas["dtime"] = pd.to_datetime(df_pandas["dtime"])
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))

    def test_xport(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"))
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))
        #self.assertTrue(meta.creation_time==datetime(2018, 8, 14, 10, 55, 46))
        #self.assertTrue(meta.modification_time==datetime(2018, 8, 14, 10, 55, 46))

    def test_xport_v5(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sas.xpt5"))
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.xptv5v8))

    def test_xport_v8(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sas.xpt8"))
        self.assertTrue(df.equals(self.xptv5v8))

    def test_xport_metaonly(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"))
        df2, meta2 = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta2.number_rows is None)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)

    def test_xport_usecols(self):
        # Currently readstat does not support skipping cols for XPT files,
        usecols = [x.upper() for x in self.usecols]
        df, meta = pyreadstat.pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"), usecols=usecols)
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.df_usecols))
        self.assertTrue(meta.number_columns == len(self.usecols))

    def test_xport_nodates(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"), disable_datetime_conversion=True)
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.df_nodates_sastata))

    def test_xport_chunks(self):
        df, meta = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "sample.xpt"), row_limit = 2, row_offset =1)
        df.columns = [x.lower() for x in df.columns]
        df_pandas = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas["dtime"] = pd.to_datetime(df_pandas["dtime"])
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))

    def test_dta(self):
        # discard dtime and arrange time
        df, meta = pyreadstat.read_dta(os.path.join(self.basic_data_folder, "sample.dta"))
        df_pandas = self.df_pandas.copy()
        df_pandas["myord"] = df_pandas["myord"].astype(np.int64)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(np.int64)
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))
        self.assertTrue(meta.readstat_variable_types["mychar"]=="string")
        self.assertTrue(meta.readstat_variable_types["myord"]=="int8")
        self.assertTrue(meta.readstat_variable_types["dtime"]=="double")
        #self.assertTrue(meta.creation_time==datetime(2018, 12, 17, 14, 53))
        #self.assertTrue(meta.modification_time==datetime(2018, 12, 17, 14, 53))

    def test_dta_metaonly(self):
        df, meta = pyreadstat.read_dta(os.path.join(self.basic_data_folder, "sample.dta"))
        df2, meta2 = pyreadstat.read_dta(os.path.join(self.basic_data_folder, "sample.dta"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta.number_rows == meta2.number_rows)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)

    def test_dta_usecols(self):
        df, meta = pyreadstat.read_dta(os.path.join(self.basic_data_folder, "sample.dta"), usecols=self.usecols)
        df_pandas = self.df_usecols.copy()
        df_pandas["myord"] = df_pandas["myord"].astype(np.int64)
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(self.usecols))
        self.assertTrue(meta.column_names == self.usecols)

    def test_dta_nodates(self):
        df, meta = pyreadstat.read_dta(os.path.join(self.basic_data_folder,"sample.dta"), disable_datetime_conversion=True)
        df_pandas = self.df_nodates_sastata
        df_pandas["myord"] = df_pandas["myord"].astype(np.int64)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(np.int64)
        df_pandas["dtime"] = df_pandas["dtime"] * 1000
        df_pandas["mytime"] = df_pandas["mytime"] * 1000
        self.assertTrue(df.equals(df_pandas))

    def test_dta_chunks(self):
        # discard dtime and arrange time
        df, meta = pyreadstat.read_dta(os.path.join(self.basic_data_folder, "sample.dta"), row_limit = 2, row_offset =1)
        df_pandas = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas["dtime"] = pd.to_datetime(df_pandas["dtime"])
        df_pandas["myord"] = df_pandas["myord"].astype(np.int64)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(np.int64)
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))

    def test_sav(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"))
        self.assertTrue(df.equals(self.df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))
        self.assertTrue(len(meta.notes)>0)
        self.assertTrue(meta.variable_display_width["mychar"]==9)
        self.assertTrue(meta.variable_storage_width["mychar"] == 8)
        self.assertTrue(meta.variable_measure["mychar"]=="nominal")
        self.assertTrue(meta.readstat_variable_types["mychar"]=="string")
        self.assertTrue(meta.readstat_variable_types["myord"]=="double")
        #self.assertTrue(meta.creation_time==datetime(2018, 8, 16, 17, 22, 33))
        #self.assertTrue(meta.modification_time==datetime(2018, 8, 16, 17, 22, 33))

    def test_sav_metaonly(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"))
        df2, meta2 = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta.number_rows == meta2.number_rows)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)
        self.assertTrue(len(meta2.notes) > 0)

    def test_sav_formatted(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), apply_value_formats=True, formats_as_category=True)
        #df.columns = self.df_pandas_formatted.columns
        self.assertTrue(df.equals(self.df_pandas_formatted))
        self.assertTrue(meta.number_columns == len(self.df_pandas_formatted.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas_formatted))
        self.assertTrue(len(meta.notes) > 0)

    def test_sav_usecols(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), usecols=self.usecols)
        self.assertTrue(df.equals(self.df_usecols))
        self.assertTrue(meta.number_columns == len(self.usecols))
        self.assertTrue(meta.column_names == self.usecols)

    def test_sav_missing(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample_missing.sav"))
        self.assertTrue(df.equals(self.df_missing_sav))
        self.assertTrue(meta.missing_ranges == {})
        df_user, meta_user = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample_missing.sav"), user_missing=True)
        self.assertTrue(df_user.equals(self.df_user_missing_sav))
        self.assertTrue(meta_user.missing_ranges['mynum'][0]=={'lo': 2000.0, 'hi': 3000.0})
        # user missing with usecols
        df_user, meta_user = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample_missing.sav"), user_missing=True, usecols=["mynum", "mylabl"])
        df_sub = self.df_user_missing_sav[["mynum", "mylabl"]]
        self.assertTrue(df_user.equals(df_sub))

    def test_sav_nodates(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), disable_datetime_conversion=True)
        #import pdb; pdb.set_trace()
        self.assertTrue(df.equals(self.df_nodates_spss))

    def test_sav_chunks(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), row_limit = 2, row_offset =1)
        df_pandas = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas["dtime"] = pd.to_datetime(df_pandas["dtime"])
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))
        self.assertTrue(len(meta.notes)>0)
        self.assertTrue(meta.variable_display_width["mychar"]==9)
        self.assertTrue(meta.variable_storage_width["mychar"] == 8)
        self.assertTrue(meta.variable_measure["mychar"]=="nominal")
    
    def test_sav_expand(self):
        src = os.path.join(self.basic_data_folder, "sample.sav")
        dst = "~/sample.sav"
        shutil.copyfile(src, os.path.expanduser(dst))
        df, meta = pyreadstat.read_sav(dst)
        os.remove(os.path.expanduser(dst))
        self.assertTrue(df.equals(self.df_pandas))

       
    def test_zsav(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"))
        self.assertTrue(df.equals(self.df_pandas))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))
        self.assertTrue(len(meta.notes) > 0)

    def test_zsav_metaonly(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"))
        df2, meta2 = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta.number_rows == meta2.number_rows)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)
        self.assertTrue(len(meta2.notes) > 0)

    def test_zsav_formatted(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"), apply_value_formats=True, formats_as_category=True)
        self.assertTrue(df.equals(self.df_pandas_formatted))
        self.assertTrue(meta.number_columns == len(self.df_pandas_formatted.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas_formatted))
        self.assertTrue(len(meta.notes) > 0)

    def test_zsav_usecols(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"), usecols=self.usecols)
        self.assertTrue(df.equals(self.df_usecols))
        self.assertTrue(meta.number_columns == len(self.usecols))
        self.assertTrue(meta.column_names == self.usecols)

    def test_zsav_nodates(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"), disable_datetime_conversion=True)
        self.assertTrue(df.equals(self.df_nodates_spss))

    def test_zsav_chunks(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.zsav"), row_limit = 2, row_offset =1)
        df_pandas = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas["dtime"] = pd.to_datetime(df_pandas["dtime"])
        self.assertTrue(df.equals(df_pandas))
        self.assertTrue(meta.number_columns == len(df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas))
        self.assertTrue(len(meta.notes)>0)
        self.assertTrue(meta.variable_display_width["mychar"]==9)
        self.assertTrue(meta.variable_storage_width["mychar"] == 8)
        self.assertTrue(meta.variable_measure["mychar"]=="nominal")

    def test_por(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"))
        df_pandas_por = self.df_pandas.copy()
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(df_pandas_por))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas_por))
        self.assertTrue(len(meta.notes) > 0)
        #self.assertTrue(meta.creation_time==datetime(2018, 12, 16, 17, 28, 21))
        #self.assertTrue(meta.modification_time==datetime(2018, 12, 16, 17, 28, 21))

    def test_por_formatted(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"), apply_value_formats=True, formats_as_category=True)
        df_pandas_por = self.df_pandas_formatted.copy()
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(df_pandas_por))
        self.assertTrue(meta.number_columns == len(self.df_pandas_formatted.columns))
        self.assertTrue(meta.number_rows == len(df_pandas_por))
        self.assertTrue(len(meta.notes) > 0)

    def test_por_metaonly(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"))
        df2, meta2 = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"), metadataonly=True)
        self.assertTrue(df2.empty)
        self.assertTrue(meta.number_columns == meta2.number_columns)
        self.assertTrue(meta2.number_rows is None)
        self.assertTrue(meta.column_names == meta2.column_names)
        self.assertTrue(meta.column_labels == meta2.column_labels)
        self.assertTrue(len(meta2.notes) > 0)

    def test_por_usecols(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"),  usecols=["MYNUM"])
        df_pandas_por = self.df_pandas[["mynum"]]
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(df_pandas_por))
        self.assertTrue(meta.number_columns == len(df_pandas_por.columns.values.tolist()))

    def test_por_nodates(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"), disable_datetime_conversion=True)
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.df_nodates_spss))

    def test_por_chunks(self):
        df, meta = pyreadstat.read_por(os.path.join(self.basic_data_folder, "sample.por"), row_limit = 2, row_offset =1)
        df_pandas_por = self.df_pandas.iloc[1:3,:].reset_index(drop=True)
        df_pandas_por['dtime'] = pd.to_datetime(df_pandas_por.dtime)
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(df_pandas_por))
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(df_pandas_por))
        self.assertTrue(len(meta.notes) > 0)

    def test_sas_catalog_win(self):
        """these sas7bdat and sasbcat where produced on windows, probably 32 bit"""
        dat = os.path.join(self.catalog_data_folder, "test_data_win.sas7bdat")
        cat = os.path.join(self.catalog_data_folder, "test_formats_win.sas7bcat")
        df, meta = pyreadstat.read_sas7bdat(dat, catalog_file=cat)
        self.assertTrue(df.equals(self.df_sas_format))

    def test_sas_catalog_linux(self):
        """these sas7bdat and sasbcat where produced on linux 64 bit"""
        dat = os.path.join(self.catalog_data_folder, "test_data_linux.sas7bdat")
        cat = os.path.join(self.catalog_data_folder, "test_formats_linux.sas7bcat")
        df, meta = pyreadstat.read_sas7bdat(dat, catalog_file=cat)
        self.assertTrue(df.equals(self.df_sas_format))

    def test_sas_dates(self):
        sas_file = os.path.join(self.basic_data_folder, "dates.sas7bdat")
        df_sas, meta = pyreadstat.read_sas7bdat(sas_file)
        self.assertTrue(df_sas.equals(self.df_sas_dates))

    def test_sas_dates_as_pandas(self):
        sas_file = os.path.join(self.basic_data_folder, "dates.sas7bdat")
        df_sas, meta = pyreadstat.read_sas7bdat(sas_file, dates_as_pandas_datetime=True)
        self.assertTrue(df_sas.equals(self.df_sas_dates_as_pandas))
        
    def test_sas_user_missing(self):
        sas_file = os.path.join(self.missing_data_folder, "missing_test.sas7bdat")
        cat_file = os.path.join(self.missing_data_folder, "missing_formats.sas7bcat")
        unformatted_csv = os.path.join(self.missing_data_folder, "missing_unformatted.csv")
        formatted_csv = os.path.join(self.missing_data_folder, "missing_sas_formatted.csv")
        labeled_csv = os.path.join(self.missing_data_folder, "missing_sas_labeled.csv")
        
        df_sas, meta = pyreadstat.read_sas7bdat(sas_file)
        df_csv = pd.read_csv(unformatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
        df_sas, meta = pyreadstat.read_sas7bdat(sas_file, user_missing=True)
        df_csv = pd.read_csv(formatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        missing_user_values = {'var1':['A'],'var2': ['B'], 'var3':['C'], 'var4':['X'], 'var5':['Y'], 
        'var6':['Z'], 'var7':['_']}
        self.assertDictEqual(meta.missing_user_values, missing_user_values)

        df_sas, meta = pyreadstat.read_sas7bdat(sas_file,
                            catalog_file=cat_file, user_missing=True,
                            formats_as_category=False)
        df_csv = pd.read_csv(labeled_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
    def test_dta_user_missing(self):
        dta_file = os.path.join(self.missing_data_folder, "missing_test.dta")
        unformatted_csv = os.path.join(self.missing_data_folder, "missing_unformatted.csv")
        formatted_csv = os.path.join(self.missing_data_folder, "missing_dta_formatted.csv")
        labeled_csv = os.path.join(self.missing_data_folder, "missing_dta_labeled.csv")
        
        df_sas, meta = pyreadstat.read_dta(dta_file)
        df_csv = pd.read_csv(unformatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
        df_sas, meta = pyreadstat.read_dta(dta_file, user_missing=True)
        df_csv = pd.read_csv(formatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        missing_user_values = {'var1':['a'],'var2': ['b'], 'var3':['c'], 'var4':['x'], 'var5':['y'], 'var6':['z']}
        self.assertDictEqual(meta.missing_user_values, missing_user_values)
        
        df_sas, meta = pyreadstat.read_dta(dta_file,
                            apply_value_formats=True, user_missing=True,
                            formats_as_category=False)
        df_csv = pd.read_csv(labeled_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
    def test_sav_user_missing(self):
        sav_file = os.path.join(self.missing_data_folder, "missing_test.sav")
        unformatted_csv = os.path.join(self.missing_data_folder, "missing_sav_unformatted.csv")
        formatted_csv = os.path.join(self.missing_data_folder, "missing_sav_formatted.csv")
        labeled_csv = os.path.join(self.missing_data_folder, "missing_sav_labeled.csv")
        
        df_sas, meta = pyreadstat.read_sav(sav_file)
        df_csv = pd.read_csv(unformatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
        df_sas, meta = pyreadstat.read_sav(sav_file, user_missing=True)
        df_csv = pd.read_csv(formatted_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
        df_sas, meta = pyreadstat.read_sav(sav_file,
                            apply_value_formats=True, user_missing=True,
                            formats_as_category=False)
        df_sas.loc[1, 'var1'] = int(df_sas['var1'][1])
        df_sas['var1'] = df_sas['var1'].astype(str)
        df_csv = pd.read_csv(labeled_csv)
        self.assertTrue(df_sas.equals(df_csv))
        
    def test_sav_missing_char(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.missing_data_folder, "missing_char.sav"))
        mdf = pd.DataFrame([[np.nan], ["a"]], columns=["mychar"])
        self.assertTrue(df.equals(mdf))
        self.assertTrue(meta.missing_ranges == {})
        df2, meta2 = pyreadstat.read_sav(os.path.join(self.missing_data_folder, "missing_char.sav"), user_missing=True)
        mdf2 = pd.DataFrame([["Z"], ["a"]], columns=["mychar"])
        self.assertTrue(df2.equals(mdf2))
        self.assertTrue(meta2.missing_ranges['mychar'][0]=={'lo': "Z", 'hi': "Z"})
    
    # test reading metadata for multiple response data

    def test_sav_multiple_response(self):
        """Assert MR data is correctly read from sav into metadata."""
        _, meta = pyreadstat.read_sav(os.path.join(self.mr_data_folder, "simple_alltypes.sav"))
        assert meta.mr_sets == {
            "categorical_array": {
                "type": "C",
                "is_dichotomy": False,
                "counted_value": None,
                "label": "",
                "variable_list": ["ca_subvar_1", "ca_subvar_2", "ca_subvar_3"]
            },
            "mymrset": {
                "type": "D",
                "is_dichotomy": True,
                "counted_value": 1,
                "label": "My multiple response set",
                "variable_list": ["bool1", "bool2", "bool3"]
            }
        }
    
    def test_sav_without_multiple_response(self):
        """Assert MR data is read as empty dict when not present in sav."""
        _, meta = pyreadstat.read_sav(os.path.join(self.missing_data_folder, "missing_char.sav"))
        assert meta.mr_sets == {}

    # read in chunks

    def test_chunk_reader(self):
        fpath = os.path.join(self.basic_data_folder, "sample.sas7bdat")
        reader = pyreadstat.read_file_in_chunks(pyreadstat.read_sas7bdat, fpath, chunksize= 2, offset=1, limit=2, disable_datetime_conversion=True)
        
        for df, meta in reader:
            pass
        
        currow = self.df_nodates_sastata.iloc[1:3,:].reset_index(drop=True)
        self.assertTrue(df.equals(currow))

    # read multiprocessing

    def test_multiprocess_reader(self):
        fpath = os.path.join(self.basic_data_folder, "sample_large.sav")
        df_multi, meta_multi = pyreadstat.read_file_multiprocessing(pyreadstat.read_sav, fpath) 
        df_single, meta_single = pyreadstat.read_sav(fpath)
        self.assertTrue(df_multi.equals(df_single))
        self.assertEqual(meta_multi.number_rows, meta_single.number_rows)

    def test_chunk_reader_multiprocess(self):
        fpath = os.path.join(self.basic_data_folder, "sample_large.sav")
        reader = pyreadstat.read_file_in_chunks(pyreadstat.read_sav, fpath, chunksize= 50, multiprocess=True)
        alldfs = list()
        for df, meta in reader:
            alldfs.append(df)
        df_multi = pd.concat(alldfs, axis=0, ignore_index=True) 
        df_single, meta_single = pyreadstat.read_sav(fpath)
        self.assertTrue(df_multi.equals(df_single))

    def test_chunk_reader_multiprocess_dict(self):
        fpath = os.path.join(self.basic_data_folder, "sample_large.sav")
        reader = pyreadstat.read_file_in_chunks(pyreadstat.read_sav, fpath, chunksize= 50, multiprocess=True, output_format='dict')
        alldfs = list()
        for chunkdict, meta in reader:
            df = pd.DataFrame(chunkdict)
            alldfs.append(df)
        df_multi = pd.concat(alldfs, axis=0, ignore_index=True) 
        df_single, meta_single = pyreadstat.read_sav(fpath)
        self.assertTrue(df_multi.equals(df_single))


    def test_multiprocess_reader_xport(self):
        fpath = os.path.join(self.basic_data_folder, "sample.xpt")
        df_multi, meta_multi = pyreadstat.read_file_multiprocessing(pyreadstat.read_xport, fpath, num_rows=1000) 
        df_single, meta_single = pyreadstat.read_xport(fpath)
        self.assertTrue(df_multi.equals(df_single))

    # writing

    def test_sav_write_basic(self):
        file_label = "basic write"
        file_note = "These are some notes"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
        missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
        #variable_alignment = {'mychar':"center", 'myord':"right"}
        variable_display_width = {'mychar':20}
        variable_measure = {"mychar": "nominal"}
        path = os.path.join(self.write_folder, "basic_write.sav")
        pyreadstat.write_sav(self.df_pandas, path, file_label=file_label, column_labels=col_labels, note=file_note, 
            variable_value_labels=variable_value_labels, missing_ranges=missing_ranges, variable_display_width=variable_display_width,
            variable_measure=variable_measure) #, variable_alignment=variable_alignment)
        df, meta = pyreadstat.read_sav(path, user_missing=True)
        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertEqual(meta.notes[0], file_note)
        self.assertDictEqual(meta.variable_value_labels, variable_value_labels)
        self.assertEqual(meta.variable_display_width['mychar'], variable_display_width['mychar'])
        #self.assertDictEqual(meta.variable_alignment, variable_alignment)
        self.assertEqual(meta.variable_measure["mychar"], variable_measure["mychar"])

    def test_sav_write_basic_expanduser(self):
        file_label = "basic write"
        file_note = "These are some notes"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
        missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
        #variable_alignment = {'mychar':"center", 'myord':"right"}
        variable_display_width = {'mychar':20}
        variable_measure = {"mychar": "nominal"}
        path = "~/sav_expand.sav" 
        pyreadstat.write_sav(self.df_pandas, path, file_label=file_label, column_labels=col_labels, note=file_note, 
            variable_value_labels=variable_value_labels, missing_ranges=missing_ranges, variable_display_width=variable_display_width,
            variable_measure=variable_measure) #, variable_alignment=variable_alignment)
        df, meta = pyreadstat.read_sav(path, user_missing=True)
        os.remove(os.path.expanduser(path))
        self.assertTrue(df.equals(self.df_pandas))

    def test_zsav_write_basic(self):
        file_label = "basic write"
        file_note = "These are some notes"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
        missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
        path = os.path.join(self.write_folder, "basic_write.zsav")
        pyreadstat.write_sav(self.df_pandas, path, file_label=file_label, column_labels=col_labels, compress=True, note=file_note,
                     variable_value_labels=variable_value_labels, missing_ranges=missing_ranges)
        df, meta = pyreadstat.read_sav(path, user_missing=True)
        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertEqual(meta.notes[0], file_note)
        self.assertDictEqual(meta.variable_value_labels, variable_value_labels)

    def test_dta_write_basic(self):
        df_pandas = self.df_pandas.copy()
        df_pandas["myord"] = df_pandas["myord"].astype(np.int32)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(np.int32)

        file_label = "basic write"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        variable_value_labels = {'mylabl': {1: 'Male', 2: 'Female'}, 'myord': {1: 'low', 2: 'medium', 3: 'high'}}
        path = os.path.join(self.write_folder, "basic_write.dta")
        pyreadstat.write_dta(df_pandas, path, file_label=file_label, column_labels=col_labels, version=12, variable_value_labels=variable_value_labels)
        df, meta = pyreadstat.read_dta(path)

        df_pandas["myord"] = df_pandas["myord"].astype(np.int64)
        df_pandas["mylabl"] = df_pandas["mylabl"].astype(np.int64)

        self.assertTrue(df.equals(df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertDictEqual(meta.variable_value_labels, variable_value_labels)

    def test_dta_write_user_missing(self):
        df_csv = pd.DataFrame([[3,"a"],["a","b"]], columns=["Var1", "Var2"])
        df_csv2 = pd.DataFrame([[3,"a"],["labeled","b"]], columns=["Var1", "Var2"])

        missing_user_values = {'Var1': ['a']}
        variable_value_labels = {'Var1':{'a':'labeled'}}
        path = os.path.join(self.write_folder, "user_missing_write.dta")
        pyreadstat.write_dta(df_csv, path, version=12, missing_user_values=missing_user_values, variable_value_labels=variable_value_labels)
        
        df_dta, meta = pyreadstat.read_dta(path, user_missing=True)
        self.assertTrue(df_csv.equals(df_dta))
        self.assertDictEqual(meta.missing_user_values, missing_user_values)
        
        df_dta2, meta2 = pyreadstat.read_dta(path, user_missing=True, apply_value_formats=True, formats_as_category=False)
        self.assertTrue(df_csv2.equals(df_dta2))

    def test_xport_write_basic_v8(self):
        file_label = "basic write"
        table_name = "TEST"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        path = os.path.join(self.write_folder, "write.xpt")
        pyreadstat.write_xport(self.df_pandas, path, file_label=file_label, column_labels=col_labels, table_name=table_name, file_format_version=8)
        df, meta = pyreadstat.read_xport(path)
        df.columns = [x.lower() for x in df.columns]

        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertEqual(table_name, meta.table_name)

    def test_xport_write_basic_v5(self):
        file_label = "basic write"
        table_name = "TEST"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        path = os.path.join(self.write_folder, "write.xpt")
        pyreadstat.write_xport(self.df_pandas, path, file_label=file_label, column_labels=col_labels, table_name=table_name, file_format_version=5)
        df, meta = pyreadstat.read_xport(path)
        df.columns = [x.lower() for x in df.columns]

        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertEqual(table_name, meta.table_name)

    def test_por_write_basic(self):

        file_label = "basic write"
        #file_note = "These are some notes"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        path = os.path.join(self.write_folder, "write.por")
        pyreadstat.write_por(self.df_pandas, path, file_label=file_label, column_labels=col_labels) #, note=file_note)
        df, meta = pyreadstat.read_por(path)
        df.columns = [x.lower() for x in df.columns]
        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        #self.assertEqual(meta.notes[0], file_note)

    def test_sav_write_dates(self):

        path = os.path.join(self.write_folder, "dates_write.sav")
        pyreadstat.write_sav(self.df_sas_dates2, path)
        df, meta = pyreadstat.read_sav(path)
        self.assertTrue(df.equals(self.df_sas_dates2))

    def test_zsav_write_dates(self):

        path = os.path.join(self.write_folder, "dates_write_zsav.sav")
        pyreadstat.write_sav(self.df_sas_dates2, path, compress=True)
        df, meta = pyreadstat.read_sav(path)
        self.assertTrue(df.equals(self.df_sas_dates2))

    def test_dta_write_dates(self):

        path = os.path.join(self.write_folder, "dates_write.dta")
        pyreadstat.write_dta(self.df_sas_dates, path)
        df, meta = pyreadstat.read_dta(path)
        self.assertTrue(df.equals(self.df_sas_dates))

    def test_xport_write_dates(self):

        path = os.path.join(self.write_folder, "dates_write.xpt")
        pyreadstat.write_xport(self.df_sas_dates2, path)
        df, meta = pyreadstat.read_xport(path)
        #import pdb;pdb.set_trace()
        self.assertTrue(df.equals(self.df_sas_dates2))

    def test_sav_write_charnan(self):
        path = os.path.join(self.write_folder, "charnan.sav")
        pyreadstat.write_sav(self.df_charnan, path)
        df, meta = pyreadstat.read_sav(path)
        df2 = self.df_charnan
        df2.iloc[0,1] = ""
        df2.iloc[0,2] = ""
        df2['integer'] = df2["integer"].astype(float)
        df2['object'] = df2['object'].astype(str)
        self.assertTrue(df2.equals(df))

    def test_zsav_write_charnan(self):
        path = os.path.join(self.write_folder, "charnan_zsav.sav")
        pyreadstat.write_sav(self.df_charnan, path, compress=True)
        df, meta = pyreadstat.read_sav(path)
        df2 = self.df_charnan
        df2.iloc[0,1] = ""
        df2.iloc[0,2] = ""
        df2['integer'] = df2["integer"].astype(float)
        df2['object'] = df2['object'].astype(str)
        self.assertTrue(df2.equals(df))

    def test_xport_write_charnan(self):
        path = os.path.join(self.write_folder, "charnan.xpt")
        pyreadstat.write_xport(self.df_charnan, path)
        df, meta = pyreadstat.read_xport(path)
        df2 = self.df_charnan
        df2.iloc[0,1] = ""
        df2.iloc[0,2] = ""
        df2['integer'] = df2["integer"].astype(float)
        df2['object'] = df2['object'].astype(str)
        self.assertTrue(df2.equals(df))

    def test_por_write_charnan(self):
        path = os.path.join(self.write_folder, "charnan_zsav.por")
        pyreadstat.write_por(self.df_charnan, path)
        df, meta = pyreadstat.read_por(path)
        df.columns = [x.lower() for x in df.columns]
        df2 = self.df_charnan
        df2.iloc[0,1] = ""
        df2.iloc[0,2] = ""
        df2['integer'] = df2["integer"].astype(float)
        df2['object'] = df2['object'].astype(str)
        self.assertTrue(df2.equals(df))

    def test_dta_write_charnan(self):
        path = os.path.join(self.write_folder, "charnan.dta")
        pyreadstat.write_dta(self.df_charnan, path)
        df, meta = pyreadstat.read_dta(path)
        df2 = self.df_charnan
        df2.iloc[0,1] = ""
        df2.iloc[0,2] = ""
        df2['integer'] = df2["integer"].astype(float)
        df2['object'] = df2['object'].astype(str)
        self.assertTrue(df2.equals(df))

    def test_set_value_labels(self):

        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"))
        df_formatted = pyreadstat.set_value_labels(df, meta, formats_as_category=True)
        #df.columns = self.df_pandas_formatted.columns
        self.assertTrue(df_formatted.equals(self.df_pandas_formatted))
        # partial
        sub1_raw = df[['myord']]
        sub1 = pyreadstat.set_value_labels(sub1_raw, meta, formats_as_category=True)
        sub2 = self.df_pandas_formatted[['myord']]
        self.assertTrue(sub1.equals(sub2))
        
    def test_update_delete_file(self):
    
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"))
        dst_path = os.path.join(self.write_folder, "update_test.sav")
        pyreadstat.write_sav(df, dst_path, variable_value_labels = meta.variable_value_labels)
        # update
        meta.variable_value_labels.update({'mylabl':{1.0:"Gents", 2.0:"Ladies"}})
        pyreadstat.write_sav(df, dst_path, variable_value_labels = meta.variable_value_labels)
        df2, meta2 = pyreadstat.read_sav(dst_path)
        self.assertDictEqual(meta2.variable_value_labels, meta.variable_value_labels)
        os.remove(dst_path)

    def test_xport_write_dates2_v8(self):
        # this sas7bdat file has features that are not compatible with v5
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "dates_xpt.sas7bdat"))
        dst_path = os.path.join(self.write_folder, "dates_xptv8.xpt")
        pyreadstat.write_xport(df, dst_path, file_format_version=8)
        df2, meta2 = pyreadstat.read_xport(dst_path)
        self.assertTrue(df.equals(df2))

    def test_xport_dates2_v8(self):
        # this sas7bdat file has features that are not compatible with v5
        df, meta = pyreadstat.read_sas7bdat(os.path.join(self.basic_data_folder, "dates_xpt.sas7bdat"))
        # this xpt file was written in SAS from the sas7bdat file
        df2, meta2 = pyreadstat.read_xport(os.path.join(self.basic_data_folder, "dates_xpt_v8.xpt"))
        self.assertTrue(df.equals(df2))
        self.assertListEqual(meta.column_labels, meta2.column_labels)

    def test_sav_international_utf8_char_value(self):
        # a file that has a value with international characters and the file is coded in utf-8
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "tegulu.sav"))
        self.assertTrue(df.iloc[0,1] == "నేను గతంలో వాడిన బ")

    def test_sav_international_varname(self):
        # a file with a varname with international characters
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "hebrews.sav"))
        self.assertTrue(df.columns[0] == "ותק_ב")

    def test_sav_original_var_types(self):
        # a file with a varname with international characters
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "test_width.sav"))
        self.assertEqual(meta.original_variable_types['StartDate'],'A1024')
        self.assertEqual(meta.original_variable_types['ResponseId'],'A18')
        self.assertEqual(meta.original_variable_types['Duration__in_seconds_'],'F40.2')
        self.assertEqual(meta.original_variable_types['Finished'],'F1.0')
        self.assertEqual(meta.readstat_variable_types['Finished'],'double')

    def test_sav_write_longstr(self):
        path = os.path.join(self.write_folder, "longstr.sav")
        pyreadstat.write_sav(self.df_longstr, path, variable_display_width={"v1": 1000})
        df, meta = pyreadstat.read_sav(path)
        self.assertTrue(meta.variable_display_width['v1']==1000)
        self.assertTrue(len(df.iloc[0,0])==781)
        
    def test_sas7bdat_file_label_linux(self):
        "testing file label for file produced on linux"
        path = os.path.join(self.basic_data_folder, "test_file_label_linux.sas7bdat")
        df, meta = pyreadstat.read_sas7bdat(path)
        self.assertEqual(meta.file_label, "mytest label")
        self.assertEqual(meta.table_name, "TEST_DATA")

    def test_sas7bdat_extra_date_formats(self):
        "testing extra date format argument"
        path = os.path.join(self.basic_data_folder, "date_test.sas7bdat")
        df, meta = pyreadstat.read_sas7bdat(path, extra_date_formats=["MMYY", "YEAR"])
        self.assertEqual(df['yr'].iloc[0], date(2023,1,1))
        self.assertEqual(df['dtc4'].iloc[0], date(2023,7,1))

    def test_sas7bdat_file_label_windows(self):
        "testing file label for file produced on windows"
        path = os.path.join(self.basic_data_folder, "test_file_label_win.sas7bdat")
        df, meta = pyreadstat.read_sas7bdat(path)
        self.assertEqual(meta.file_label, "mytest label")
        self.assertEqual(meta.table_name, "TEST_DATA")

    def test_sav_write_variable_formats(self):
        "testing variable formats for SAV files"
        path = os.path.join(self.write_folder, "variable_format.sav")
        df = pd.DataFrame({'restricted':[1023, 10], 'integer':[1,2]})
        formats = {'restricted':'restricted_integer', 'integer':'integer'}
        pyreadstat.write_sav(df, path, variable_format=formats)
        df2, meta2 = pyreadstat.read_sav(path)
        self.assertEqual(meta2.original_variable_types['restricted'], "N4")
        self.assertEqual(meta2.original_variable_types['integer'], "F1.0")

    def test_sav_ordered_categories(self):
        path = os.path.join(self.basic_data_folder, "ordered_category.sav")
        df, meta = pyreadstat.read_sav(path, apply_value_formats=True, formats_as_ordered_category=True)
        self.assertTrue(df.Col1.cat.ordered)
        self.assertListEqual(list(df.Col1.cat.categories), ['high', 'low', 'medium'])

    def test_sav_pathlib(self):
        if is_pathlib_available:
            path = Path(self.basic_data_folder).joinpath("sample.sav")
            df, meta = pyreadstat.read_sav(path)
            self.assertTrue(df.equals(self.df_pandas))

    def test_sav_write_pathlib(self):
        if is_pathlib_available:
            file_label = "basic write"
            file_note = "These are some notes"
            col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
            variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
            missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
            #variable_alignment = {'mychar':"center", 'myord':"right"}
            variable_display_width = {'mychar':20}
            variable_measure = {"mychar": "nominal"}
            path = Path(self.write_folder).joinpath('pathlib_write.sav')
            pyreadstat.write_sav(self.df_pandas, path, file_label=file_label, column_labels=col_labels, note=file_note, 
                variable_value_labels=variable_value_labels, missing_ranges=missing_ranges, variable_display_width=variable_display_width,
                variable_measure=variable_measure) #, variable_alignment=variable_alignment)
            df, meta = pyreadstat.read_sav(path, user_missing=True)
            self.assertTrue(df.equals(self.df_pandas))
            self.assertEqual(meta.file_label, file_label)
            self.assertListEqual(meta.column_labels, col_labels)
            self.assertEqual(meta.notes[0], file_note)
            self.assertDictEqual(meta.variable_value_labels, variable_value_labels)
            self.assertEqual(meta.variable_display_width['mychar'], variable_display_width['mychar'])
            #self.assertDictEqual(meta.variable_alignment, variable_alignment)
            self.assertEqual(meta.variable_measure["mychar"], variable_measure["mychar"])

    def test_sav_write_dictlabels(self):
        col_names = ["mychar", "mynum", "mydate", "dtime", "mylabl", "myord", "mytime"]
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        col_lab_dict = {k:v for k,v in zip(col_names, col_labels) if v}
        variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
        missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
        path = os.path.join(self.write_folder, "dictlabel_write.sav")
        pyreadstat.write_sav(self.df_pandas, path, column_labels=col_lab_dict)
        df, meta = pyreadstat.read_sav(path, user_missing=True)
        self.assertTrue(df.equals(self.df_pandas))
        self.assertListEqual(meta.column_labels, col_labels)
    
    def test_dta_write_single_value_user_missing(self):
        df = pd.DataFrame({"var": ["a", "a", "a", "a"]})
        missing_user_values = {"var": ["a"]}
        path = os.path.join(self.write_folder, "singleusermissing.dta")
        pyreadstat.write_dta(df=df, dst_path=path, missing_user_values=missing_user_values,version=12) 
        df2, meta2 = pyreadstat.read_dta(path, user_missing=True)
        self.assertTrue(df.equals(df2))

    def test_dta_write_only_missing_and_user_missing(self):
        df = pd.DataFrame({"var": [np.nan, "a", "b"]})
        path = os.path.join(self.write_folder, "onlymissing_and_usermissing.dta")
        variable_value_labels={"var": {1: "Val 1", 2: "Val 2", 3: "Val 3", "a": "Missing A", "b": "Missing B", } }
        missing_user_values={"var": ["a", "b"]}
        pyreadstat.write_dta(df, path, variable_value_labels=variable_value_labels, missing_user_values=missing_user_values,version=12)
        df2, meta2 = pyreadstat.read_dta(path, user_missing=True, )
        self.assertTrue(df.equals(df2))

    def test_sav_outputformat_dict(self):
        df, meta = pyreadstat.read_sav(os.path.join(self.basic_data_folder, "sample.sav"), output_format='dict')
        self.assertTrue(meta.number_columns == len(self.df_pandas.columns))
        self.assertTrue(meta.number_rows == len(self.df_pandas))
        self.assertTrue(len(meta.notes)>0)
        self.assertTrue(meta.variable_display_width["mychar"]==9)
        self.assertTrue(meta.variable_storage_width["mychar"] == 8)
        self.assertTrue(meta.variable_measure["mychar"]=="nominal")
        self.assertTrue(meta.readstat_variable_types["mychar"]=="string")
        self.assertTrue(meta.readstat_variable_types["myord"]=="double")
        padic = self.df_pandas.to_dict(orient='list')
        #import pdb;pdb.set_trace()
        for colname, data in df.items():
            curdfcol = df[colname]
            for indx, val in enumerate(data):
                if pd.isna(val) and pd.isna(curdfcol[indx]):
                    continue
                self.assertTrue(val==curdfcol[indx])

    def test_sav_write_rowcompression(self):
        file_label = "row compression write"
        file_note = "These are some notes"
        col_labels = ["mychar label","mynum label", "mydate label", "dtime label", None, "myord label", "mytime label"]
        variable_value_labels = {'mylabl': {1.0: 'Male', 2.0: 'Female'}, 'myord': {1.0: 'low', 2.0: 'medium', 3.0: 'high'}}
        missing_ranges = {'mychar':['a'], 'myord': [{'hi':2, 'lo':1}]}
        #variable_alignment = {'mychar':"center", 'myord':"right"}
        variable_display_width = {'mychar':20}
        variable_measure = {"mychar": "nominal"}
        path = os.path.join(self.write_folder, "rowcompression_write.sav")
        pyreadstat.write_sav(self.df_pandas, path, file_label=file_label, column_labels=col_labels, note=file_note, 
            variable_value_labels=variable_value_labels, missing_ranges=missing_ranges, variable_display_width=variable_display_width,
            variable_measure=variable_measure, row_compress=True) #, variable_alignment=variable_alignment)
        df, meta = pyreadstat.read_sav(path, user_missing=True)
        self.assertTrue(df.equals(self.df_pandas))
        self.assertEqual(meta.file_label, file_label)
        self.assertListEqual(meta.column_labels, col_labels)
        self.assertEqual(meta.notes[0], file_note)
        self.assertDictEqual(meta.variable_value_labels, variable_value_labels)
        self.assertEqual(meta.variable_display_width['mychar'], variable_display_width['mychar'])
        #self.assertDictEqual(meta.variable_alignment, variable_alignment)
        self.assertEqual(meta.variable_measure["mychar"], variable_measure["mychar"])

if __name__ == '__main__':

    import sys

    if "--inplace" in sys.argv:

        script_folder = os.path.split(os.path.split(os.path.realpath(__file__))[0])[0]
        sys.path.insert(0, script_folder)
        sys.argv.remove('--inplace')

    import pyreadstat

    print("package location:", pyreadstat.__file__)

    unittest.main()