Revision b0d884e9 snf-tools/synnefo_tools/burnin/pithos_tests.py

b/snf-tools/synnefo_tools/burnin/pithos_tests.py
639 639
        self.assertTrue('x-object-public' in r.headers)
640 640
        self.info('Publishing works')
641 641

  
642
        vers2 = int(r.headers['x-object-version'])
643 642
        etag = r.headers['etag']
644 643
        self.assertEqual(r.text, 'b')
645 644
        self.info('Remote object content is correct')
......
706 705
            content_encoding='application/octet-stream',
707 706
            source_account='nonExistendAddress@NeverLand.com',
708 707
            content_length=0,
709
            success=(201, 403))
710
        self.assertEqual(r.status_code, 403)
708
            success=(403, ))
711 709
        self.info('Fake source account is handled correctly')
712 710

  
713 711
        r1 = pithos.get_object_info(obj)
......
721 719
        self.assertEqual(r1['x-object-hash'], r0['x-object-hash'])
722 720
        self.info('Cross container move is OK')
723 721

  
722
        pithos.container = self.temp_containers[-1]
723
        pithos.create_container(versioning='auto')
724
        pithos.upload_from_string(obj, 'first version')
725
        source_hashmap = pithos.get_object_hashmap(obj)['hashes']
726
        pithos.upload_from_string(obj, 'second version')
727
        pithos.upload_from_string(obj, 'third version')
728
        versions = pithos.get_object_versionlist(obj)
729
        self.assertEqual(len(versions), 3)
730
        vers0 = versions[0][0]
731

  
732
        pithos.container = self.temp_containers[-2]
724 733
        pithos.object_put(
725 734
            obj,
726 735
            format=None,
727
            move_from='/%s/dir/%s' % (self.temp_containers[-1], obj),
728
            source_version=vers2,
736
            move_from='/%s/%s' % (self.temp_containers[-1], obj),
737
            source_version=vers0,
729 738
            content_encoding='application/octet-stream',
730 739
            content_length=0, success=201)
740
        target_hashmap = pithos.get_object_hashmap(obj)['hashes']
741
        self.assertEqual(source_hashmap, target_hashmap)
731 742
        self.info('Source-version is OK')
732 743

  
733 744
        mobj = 'manifest.test'
......
758 769

  
759 770
        """MISSING: test transfer-encoding?"""
760 771

  
772
    def test_060_object_copy(self):
773
        pithos = self.clients.pithos
774
        obj, trg = 'source.file2copy', 'copied.file'
775
        data = '{"key1":"val1", "key2":"val2"}'
776

  
777
        r = pithos.object_put(
778
            obj,
779
            content_type='application/octet-stream',
780
            data=data,
781
            metadata=dict(mkey1='mval1', mkey2='mval2'),
782
            permissions=dict(
783
                read=['accX:groupA', 'u1', 'u2'],
784
                write=['u2', 'u3']),
785
            content_disposition='attachment; filename="fname.ext"')
786
        self.info('Prepared a file /%s/%s' % (pithos.container, obj))
787

  
788
        r = pithos.object_copy(
789
            obj,
790
            destination='/%s/%s' % (pithos.container, trg),
791
            ignore_content_type=False, content_type='application/json',
792
            metadata={'mkey2': 'mval2a', 'mkey3': 'mval3'},
793
            permissions={'write': ['u5', 'accX:groupB']})
794
        self.assertEqual(r.status_code, 201)
795
        self.info('Status code is OK')
796

  
797
        r = pithos.get_object_info(trg)
798
        self.assertTrue('content-disposition' in r)
799
        self.info('Content-disposition is OK')
800

  
801
        self.assertEqual(r['x-object-meta-mkey1'], 'mval1')
802
        self.assertEqual(r['x-object-meta-mkey2'], 'mval2a')
803
        self.assertEqual(r['x-object-meta-mkey3'], 'mval3')
804
        self.info('Metadata are OK')
805

  
806
        r = pithos.get_object_sharing(trg)
807
        self.assertFalse('read' in r or 'u2' in r['write'])
808
        self.assertTrue('accx:groupb' in r['write'])
809
        self.info('Sharing is OK')
810

  
811
        r = pithos.object_copy(
812
            obj,
813
            destination='/%s/%s' % (pithos.container, obj),
814
            content_encoding='utf8',
815
            content_type='application/json',
816
            destination_account='nonExistendAddress@NeverLand.com',
817
            success=(201, 404))
818
        self.assertEqual(r.status_code, 404)
819
        self.info('Non existing UUID correctly causes a 404')
820

  
821
        """Check destination being another container
822
        and also content_type and content encoding"""
823
        r = pithos.object_copy(
824
            obj,
825
            destination='/%s/%s' % (self.temp_containers[-1], obj),
826
            content_encoding='utf8',
827
            content_type='application/json')
828
        self.assertEqual(r.status_code, 201)
829
        self.assertEqual(
830
            r.headers['content-type'],
831
            'application/json; charset=UTF-8')
832

  
833
        """Check ignore_content_type and content_type"""
834
        pithos.container = self.temp_containers[-1]
835
        r = pithos.object_get(obj)
836
        etag = r.headers['etag']
837
        ctype = r.headers['content-type']
838
        self.assertEqual(ctype, 'application/json')
839
        self.info('Cross container copy w. content-type/encoding is OK')
840

  
841
        r = pithos.object_copy(
842
            obj,
843
            destination='/%s/%s0' % (pithos.container, obj),
844
            ignore_content_type=True,
845
            content_type='text/x-python')
846
        self.assertEqual(r.status_code, 201)
847
        self.assertNotEqual(r.headers['content-type'], 'application/json')
848
        r = pithos.object_get(obj + '0')
849
        self.assertNotEqual(r.headers['content-type'], 'text/x-python')
850

  
851
        r = pithos.object_copy(
852
            obj,
853
            destination='/%s/%s1' % (pithos.container, obj),
854
            if_etag_match=etag)
855
        self.assertEqual(r.status_code, 201)
856
        self.info('if-etag-match is OK')
857

  
858
        r = pithos.object_copy(
859
            obj,
860
            destination='/%s/%s2' % (pithos.container, obj),
861
            if_etag_not_match='lalala')
862
        self.assertEqual(r.status_code, 201)
863
        self.info('if-etag-not-match is OK')
864

  
865
        r = pithos.object_copy(
866
            '%s2' % obj,
867
            destination='/%s/%s3' % (pithos.container, obj),
868
            format='xml',
869
            public=True)
870
        self.assertEqual(r.status_code, 201)
871
        self.assertTrue('xml' in r.headers['content-type'])
872

  
873
        r = pithos.get_object_info(obj + '3')
874
        self.assertTrue('x-object-public' in r)
875
        self.info('Publish, format and source-version are OK')
876

  
761 877
    @classmethod
762 878
    def tearDownClass(cls):  # noqa
763 879
        """Clean up"""

Also available in: Unified diff